diff --git a/Makefile b/Makefile index efe836a21..5f015981d 100644 --- a/Makefile +++ b/Makefile @@ -575,9 +575,10 @@ local-stop: ## Stop every local stack — SubmitQueue, Stovepipe, and Runway local-stovepipe-debug-start: build-stovepipe-linux-debug ## Start Stovepipe under delve in Docker (attach IDE to :2345) @echo "Starting Stovepipe service with compose (debug)..." - @$(COMPOSE) -f $(STOVEPIPE_COMPOSE_FILE) -f $(STOVEPIPE_DEBUG_COMPOSE_FILE) -p $(STOVEPIPE_LOCAL_PROJECT) up -d --build --wait + @$(COMPOSE) -f $(STOVEPIPE_COMPOSE_FILE) -f $(STOVEPIPE_DEBUG_COMPOSE_FILE) -p $(STOVEPIPE_LOCAL_PROJECT) up -d --build --wait mysql-app mysql-queue @echo "Applying storage and queue schemas..." @$(MAKE) -s local-init-stovepipe-schemas + @$(COMPOSE) -f $(STOVEPIPE_COMPOSE_FILE) -f $(STOVEPIPE_DEBUG_COMPOSE_FILE) -p $(STOVEPIPE_LOCAL_PROJECT) up -d --build --wait @echo "" @echo "✅ Stovepipe debug is running (delve :2345)!" @echo "" @@ -591,9 +592,10 @@ local-stovepipe-logs: ## View logs from the running Stovepipe service local-stovepipe-start: build-stovepipe-linux ## Start Stovepipe service (gRPC service + MySQL storage + MySQL queue) @echo "Starting Stovepipe service with compose..." - @$(COMPOSE) -f $(STOVEPIPE_COMPOSE_FILE) -p $(STOVEPIPE_LOCAL_PROJECT) up -d --build --wait + @$(COMPOSE) -f $(STOVEPIPE_COMPOSE_FILE) -p $(STOVEPIPE_LOCAL_PROJECT) up -d --build --wait mysql-app mysql-queue @echo "Applying storage and queue schemas..." @$(MAKE) -s local-init-stovepipe-schemas + @$(COMPOSE) -f $(STOVEPIPE_COMPOSE_FILE) -p $(STOVEPIPE_LOCAL_PROJECT) up -d --build --wait @echo "" @echo "✅ Stovepipe service is running!" @echo "" diff --git a/service/stovepipe/docker-compose.yml b/service/stovepipe/docker-compose.yml index 91b5e0e0d..035f5ea0f 100644 --- a/service/stovepipe/docker-compose.yml +++ b/service/stovepipe/docker-compose.yml @@ -14,7 +14,7 @@ # Quick start: # make local-stovepipe-start # -# After `up`, the storage and queue schemas are applied (local-init-stovepipe-schemas). +# Start the databases, apply schemas (local-init-stovepipe-schemas), then start the service. services: # Storage database - Stovepipe's request and request_uri tables. diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index 90960f92e..581a1c9f5 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/service/stovepipe/server/BUILD.bazel @@ -31,6 +31,8 @@ go_library( "//stovepipe/controller/process:go_default_library", "//stovepipe/controller/record:go_default_library", "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/core/queuelog:go_default_library", + "//stovepipe/core/queuestate:go_default_library", "//stovepipe/core/requestlog:go_default_library", "//stovepipe/extension/buildrunner:go_default_library", "//stovepipe/extension/buildrunner/fake:go_default_library", diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 34589d671..b59a3ece2 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -53,6 +53,8 @@ import ( "github.com/uber/submitqueue/stovepipe/controller/process" "github.com/uber/submitqueue/stovepipe/controller/record" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/core/queuelog" + "github.com/uber/submitqueue/stovepipe/core/queuestate" "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/extension/buildrunner" buildrunnerfake "github.com/uber/submitqueue/stovepipe/extension/buildrunner/fake" @@ -245,6 +247,12 @@ func run() error { brf := fakeBuildRunnerFactory{} storageFty := storageFactory{backend: store} + stateTransitions := queuestate.NewTransitionController(storageFty, sourceControl, queuelog.NewMaterializer(scope)) + for _, queue := range tenants { + if _, err := stateTransitions.Initialize(ctx, queue); err != nil { + return fmt.Errorf("initialize queue enablement %q: %w", queue, err) + } + } materializer := requestlog.NewMaterializer(scope) primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, materializer, registry, sourceControl, brf, hookResolver{}) if err != nil { diff --git a/stovepipe/core/queuelog/BUILD.bazel b/stovepipe/core/queuelog/BUILD.bazel new file mode 100644 index 000000000..865ecbeb5 --- /dev/null +++ b/stovepipe/core/queuelog/BUILD.bazel @@ -0,0 +1,35 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = [ + "materializer.go", + "queue_summary.go", + ], + importpath = "github.com/uber/submitqueue/stovepipe/core/queuelog", + visibility = [ + "//service/stovepipe:__subpackages__", + "//stovepipe:__subpackages__", + "//test:__subpackages__", + ], + deps = [ + "//platform/metrics:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["materializer_test.go"], + embed = [":go_default_library"], + deps = [ + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "//stovepipe/extension/storage/mock:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + ], +) diff --git a/stovepipe/core/queuelog/materializer.go b/stovepipe/core/queuelog/materializer.go new file mode 100644 index 000000000..2bb79269f --- /dev/null +++ b/stovepipe/core/queuelog/materializer.go @@ -0,0 +1,125 @@ +// 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 queuelog retains queue history and projects its current status summary. +package queuelog + +import ( + "context" + "errors" + "fmt" + "math" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +// ErrInconsistentHistory identifies missing or conflicting queue history or summaries. +var ErrInconsistentHistory = errors.New("inconsistent queue history") + +// Materializer retains committed state changes and updates their summary. +type Materializer interface { + // PersistLog idempotently retains an occurrence, then projects its state. + PersistLog(context.Context, storage.Storage, entity.QueueLog) error +} + +type materializer struct{ scope tally.Scope } + +func NewMaterializer(scope tally.Scope) Materializer { + return &materializer{scope: scope.SubScope("queue_log_materializer")} +} + +func NewQueueStateLog(queue entity.Queue) entity.QueueLog { + return entity.QueueLog{ID: entity.QueueLogID(queue.EnablementVersion), Queue: queue.Name, + State: queue.Enablement, QueueStateVersion: queue.EnablementVersion, + EffectiveFromCommitURI: queue.EffectiveFromCommitURI, TimestampMs: queue.EnablementChangedAtMs} +} + +func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, log entity.QueueLog) (retErr error) { + op := metrics.Begin(m.scope, "persist_log", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...) + defer func() { op.Complete(retErr) }() + if err := ctx.Err(); err != nil { + return err + } + if err := log.Validate(); err != nil { + return fmt.Errorf("queue %q: invalid log %q at state version %d: %w: %v", log.Queue, log.ID, log.QueueStateVersion, ErrInconsistentHistory, err) + } + logs := stores.GetQueueLogStore() + if err := logs.Create(ctx, log); err != nil { + if !errors.Is(err, storage.ErrAlreadyExists) { + return fmt.Errorf("retain queue log: %w", err) + } + retained, err := logs.Get(ctx, log.ID) + if err != nil { + return fmt.Errorf("reconcile queue log: %w", err) + } + if retained != log { + return fmt.Errorf("queue %q: log %q at state version %d conflicts with retained log %q for queue %q at state version %d: %w", log.Queue, log.ID, log.QueueStateVersion, retained.ID, retained.Queue, retained.QueueStateVersion, ErrInconsistentHistory) + } + log = retained + } + return materializeQueueSummary(ctx, stores.GetQueueSummaryStore(), log) +} + +func materializeQueueSummary(ctx context.Context, store storage.QueueSummaryStore, log entity.QueueLog) error { + for { + if err := ctx.Err(); err != nil { + return err + } + current, err := store.Get(ctx) + if storage.IsNotFound(err) { + initial := entity.QueueSummary{Queue: log.Queue, State: log.State, + QueueStateVersion: log.QueueStateVersion, EffectiveFromCommitURI: log.EffectiveFromCommitURI, + StateTimestampMs: log.TimestampMs, Version: 1} + if err := store.Create(ctx, initial); errors.Is(err, storage.ErrAlreadyExists) { + continue + } else { + return err + } + } + if err != nil { + return err + } + if current.Queue != log.Queue { + return fmt.Errorf("queue %q log %q: summary belongs to queue %q at projection version %d: %w", log.Queue, log.ID, current.Queue, current.Version, ErrInconsistentHistory) + } + if err := current.Validate(); err != nil { + return fmt.Errorf("queue %q: invalid summary at projection version %d, state version %d: %w: %v", log.Queue, current.Version, current.QueueStateVersion, ErrInconsistentHistory, err) + } + if log.QueueStateVersion < current.QueueStateVersion { + return nil + } + if log.QueueStateVersion == current.QueueStateVersion { + if current.Snapshot() != log.Snapshot() { + return fmt.Errorf("queue %q: log %q conflicts with summary state version %d (projection version %d): %w", log.Queue, log.ID, current.QueueStateVersion, current.Version, ErrInconsistentHistory) + } + return nil + } + if current.Version == math.MaxInt64 { + return fmt.Errorf("queue %q log %q: summary projection version %d exhausted: %w", log.Queue, log.ID, current.Version, ErrInconsistentHistory) + } + updated := current + updated.State, updated.QueueStateVersion = log.State, log.QueueStateVersion + updated.EffectiveFromCommitURI, updated.StateTimestampMs = log.EffectiveFromCommitURI, log.TimestampMs + newVersion := current.Version + 1 + if err := store.Update(ctx, updated, current.Version, newVersion); errors.Is(err, storage.ErrVersionMismatch) { + continue + } else if err != nil { + return err + } + return nil + } +} diff --git a/stovepipe/core/queuelog/materializer_test.go b/stovepipe/core/queuelog/materializer_test.go new file mode 100644 index 000000000..707adfea5 --- /dev/null +++ b/stovepipe/core/queuelog/materializer_test.go @@ -0,0 +1,192 @@ +// 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 queuelog + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" + "go.uber.org/mock/gomock" +) + +type materializerFixture struct { + m Materializer + stores *storagemock.MockStorage + logs map[string]entity.QueueLog + summary entity.QueueSummary + logErr, summaryErr error + beforeCreate, beforeUpdate func() +} + +func newMaterializerFixture(t *testing.T) *materializerFixture { + t.Helper() + ctrl := gomock.NewController(t) + f := &materializerFixture{m: NewMaterializer(tally.NoopScope), stores: storagemock.NewMockStorage(ctrl), logs: make(map[string]entity.QueueLog)} + logs, summaries := storagemock.NewMockQueueLogStore(ctrl), storagemock.NewMockQueueSummaryStore(ctrl) + f.stores.EXPECT().GetQueueLogStore().Return(logs).AnyTimes() + f.stores.EXPECT().GetQueueSummaryStore().Return(summaries).AnyTimes() + logs.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, log entity.QueueLog) error { + if f.logErr != nil { + return f.logErr + } + if _, exists := f.logs[log.ID]; exists { + return storage.ErrAlreadyExists + } + f.logs[log.ID] = log + return nil + }).AnyTimes() + logs.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, id string) (entity.QueueLog, error) { + log, ok := f.logs[id] + if !ok { + return entity.QueueLog{}, storage.ErrNotFound + } + return log, nil + }).AnyTimes() + summaries.EXPECT().Get(gomock.Any()).DoAndReturn(func(context.Context) (entity.QueueSummary, error) { + if f.summary.Version == 0 { + return entity.QueueSummary{}, storage.ErrNotFound + } + return f.summary, nil + }).AnyTimes() + summaries.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.QueueSummary) error { + if f.beforeCreate != nil { + hook := f.beforeCreate + f.beforeCreate = nil + hook() + } + if f.summaryErr != nil { + return f.summaryErr + } + if f.summary.Version != 0 { + return storage.ErrAlreadyExists + } + require.Equal(t, summary.Snapshot(), f.logs[entity.QueueLogID(summary.QueueStateVersion)].Snapshot()) + f.summary = summary + return nil + }).AnyTimes() + summaries.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.QueueSummary, oldVersion, newVersion int64) error { + if f.beforeUpdate != nil { + hook := f.beforeUpdate + f.beforeUpdate = nil + hook() + } + if f.summaryErr != nil { + return f.summaryErr + } + if f.summary.Version != oldVersion { + return storage.ErrVersionMismatch + } + require.Equal(t, oldVersion, summary.Version) + require.Equal(t, summary.Snapshot(), f.logs[entity.QueueLogID(summary.QueueStateVersion)].Snapshot()) + summary.Version = newVersion + f.summary = summary + return nil + }).AnyTimes() + return f +} + +func testQueueLog(version int64) entity.QueueLog { + log := entity.QueueLog{ID: entity.QueueLogID(version), Queue: "repo/main", TimestampMs: version * 1000, QueueStateVersion: version, State: entity.QueueEnablementDisabled} + if version > 1 { + log.EffectiveFromCommitURI = "A" + } + if version%2 == 0 { + log.State = entity.QueueEnablementEnabled + } + return log +} + +func TestMaterializerReplayDoesNotRegressSummary(t *testing.T) { + f := newMaterializerFixture(t) + for _, v := range []int64{1, 2, 3, 2, 1, 3} { + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, testQueueLog(v))) + } + require.Len(t, f.logs, 3) + require.Equal(t, testQueueLog(3).Snapshot(), f.summary.Snapshot()) + require.Equal(t, int64(3), f.summary.Version) +} + +func TestMaterializerRecoversAfterProjectionFailure(t *testing.T) { + for _, version := range []int64{1, 2} { + t.Run(map[int64]string{1: "creation", 2: "update"}[version], func(t *testing.T) { + f := newMaterializerFixture(t) + if version == 2 { + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, testQueueLog(1))) + } + log := testQueueLog(version) + f.summaryErr = errors.New("projection unavailable") + require.ErrorIs(t, f.m.PersistLog(context.Background(), f.stores, log), f.summaryErr) + require.Equal(t, log, f.logs[log.ID]) + require.Equal(t, version-1, f.summary.QueueStateVersion) + f.summaryErr = nil + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, log)) + require.Equal(t, log.Snapshot(), f.summary.Snapshot()) + }) + } +} + +func TestMaterializerConvergesAfterProjectionRaces(t *testing.T) { + for _, creation := range []bool{true, false} { + t.Run(map[bool]string{true: "create race", false: "update race"}[creation], func(t *testing.T) { + f := newMaterializerFixture(t) + if creation { + f.beforeCreate = func() { require.NoError(t, f.m.PersistLog(context.Background(), f.stores, testQueueLog(3))) } + } else { + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, testQueueLog(1))) + f.beforeUpdate = func() { require.NoError(t, f.m.PersistLog(context.Background(), f.stores, testQueueLog(3))) } + } + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, testQueueLog(2))) + require.Equal(t, testQueueLog(3).Snapshot(), f.summary.Snapshot()) + }) + } +} + +func TestMaterializerConflictsAndStorageErrors(t *testing.T) { + for _, tc := range []struct { + name string + corrupt func(*materializerFixture) + want error + }{ + {"conflicting retained log", func(f *materializerFixture) { + log := f.logs[entity.QueueLogID(2)] + log.TimestampMs++ + f.logs[log.ID] = log + }, ErrInconsistentHistory}, + {"conflicting summary", func(f *materializerFixture) { f.summary.StateTimestampMs++ }, ErrInconsistentHistory}, + {"wrong queue summary", func(f *materializerFixture) { f.summary.Queue = "other" }, ErrInconsistentHistory}, + {"invalid summary version", func(f *materializerFixture) { f.summary.Version = -1 }, ErrInconsistentHistory}, + } { + t.Run(tc.name, func(t *testing.T) { + f := newMaterializerFixture(t) + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, testQueueLog(2))) + tc.corrupt(f) + require.ErrorIs(t, f.m.PersistLog(context.Background(), f.stores, testQueueLog(2)), tc.want) + }) + } + f := newMaterializerFixture(t) + f.logErr = errors.New("history unavailable") + require.ErrorIs(t, f.m.PersistLog(context.Background(), f.stores, testQueueLog(1)), f.logErr) + require.Empty(t, f.logs) + require.Zero(t, f.summary.Version) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + require.ErrorIs(t, f.m.PersistLog(ctx, f.stores, testQueueLog(1)), context.Canceled) +} diff --git a/stovepipe/core/queuelog/queue_summary.go b/stovepipe/core/queuelog/queue_summary.go new file mode 100644 index 000000000..6926f3e81 --- /dev/null +++ b/stovepipe/core/queuelog/queue_summary.go @@ -0,0 +1,67 @@ +// 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 queuelog + +import ( + "context" + "fmt" + + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +// LoadSummary requires the summary to match its retained history entry. +func LoadSummary(ctx context.Context, stores storage.Storage, queue string) (entity.QueueState, error) { + if err := ctx.Err(); err != nil { + return entity.QueueState{}, err + } + summary, err := stores.GetQueueSummaryStore().Get(ctx) + if err != nil { + return entity.QueueState{}, err + } + if summary.Queue != queue { + return entity.QueueState{}, fmt.Errorf("queue %q: summary belongs to queue %q at projection version %d: %w", queue, summary.Queue, summary.Version, ErrInconsistentHistory) + } + if err := summary.Validate(); err != nil { + return entity.QueueState{}, fmt.Errorf("queue %q: invalid summary at projection version %d, state version %d: %w: %v", queue, summary.Version, summary.QueueStateVersion, ErrInconsistentHistory, err) + } + retained, err := LoadVersion(ctx, stores.GetQueueLogStore(), queue, summary.QueueStateVersion) + if err != nil { + return entity.QueueState{}, err + } + if retained != summary.Snapshot() { + return entity.QueueState{}, fmt.Errorf("queue %q: summary state version %d at projection version %d conflicts with retained history: %w", queue, summary.QueueStateVersion, summary.Version, ErrInconsistentHistory) + } + return summary.Snapshot(), nil +} + +// LoadVersion returns ErrInconsistentHistory for missing or invalid history. +func LoadVersion(ctx context.Context, logs storage.QueueLogStore, queue string, version int64) (entity.QueueState, error) { + logID := entity.QueueLogID(version) + log, err := logs.Get(ctx, logID) + if err != nil { + if storage.IsNotFound(err) { + return entity.QueueState{}, fmt.Errorf("queue %q: missing log %q for state version %d: %w", queue, logID, version, ErrInconsistentHistory) + } + return entity.QueueState{}, err + } + if log.Queue != queue || log.QueueStateVersion != version { + return entity.QueueState{}, fmt.Errorf("queue %q: log %q for state version %d belongs to queue %q at state version %d: %w", queue, logID, version, log.Queue, log.QueueStateVersion, ErrInconsistentHistory) + } + if err := log.Validate(); err != nil { + return entity.QueueState{}, fmt.Errorf("queue %q: invalid log %q at state version %d: %w: %v", queue, logID, version, ErrInconsistentHistory, err) + } + return log.Snapshot(), nil +} diff --git a/stovepipe/core/queuestate/BUILD.bazel b/stovepipe/core/queuestate/BUILD.bazel new file mode 100644 index 000000000..0312d871d --- /dev/null +++ b/stovepipe/core/queuestate/BUILD.bazel @@ -0,0 +1,33 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["queuestate.go"], + importpath = "github.com/uber/submitqueue/stovepipe/core/queuestate", + visibility = [ + "//service/stovepipe:__subpackages__", + "//stovepipe:__subpackages__", + "//test:__subpackages__", + ], + deps = [ + "//stovepipe/core/queuelog:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/sourcecontrol:go_default_library", + "//stovepipe/extension/storage:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["queuestate_test.go"], + embed = [":go_default_library"], + deps = [ + "//stovepipe/core/queuelog:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/sourcecontrol:go_default_library", + "//stovepipe/extension/sourcecontrol/fake:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + ], +) diff --git a/stovepipe/core/queuestate/queuestate.go b/stovepipe/core/queuestate/queuestate.go new file mode 100644 index 000000000..852b6862d --- /dev/null +++ b/stovepipe/core/queuestate/queuestate.go @@ -0,0 +1,315 @@ +// 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 queuestate applies enablement to internal queue snapshots and repairs their read models. +package queuestate + +import ( + "context" + "errors" + "fmt" + "math" + "time" + + "github.com/uber/submitqueue/stovepipe/core/queuelog" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +// ErrInvalidChange identifies an invalid enablement decision or boundary. +var ErrInvalidChange = errors.New("invalid queue state change") + +// PreparedChange captures one decision and its immutable boundary for application retries. +type PreparedChange struct { + previous entity.QueueState + next entity.QueueState +} + +// TransitionController changes internal queue enablement and repairs its history and summary. +type TransitionController interface { + // Initialize explicitly registers disabled enablement while preserving existing state. + Initialize(context.Context, string) (entity.QueueState, error) + // Prepare pins the enablement sequence and boundary; same-state decisions are no-ops. + Prepare(context.Context, string, entity.QueueEnablement) (PreparedChange, error) + // Apply commits a prepared change; competing stale decisions are never rebased. + Apply(context.Context, PreparedChange) (entity.QueueState, error) +} + +type transitionController struct { + stores storage.Factory + sourceControl sourcecontrol.Factory + materializer queuelog.Materializer + now func() time.Time +} + +func NewTransitionController(stores storage.Factory, sc sourcecontrol.Factory, materializer queuelog.Materializer) TransitionController { + return &transitionController{stores: stores, sourceControl: sc, materializer: materializer, now: time.Now} +} + +func (c *transitionController) Initialize(ctx context.Context, name string) (entity.QueueState, error) { + stores, err := c.resolveStores(ctx, name) + if err != nil { + return entity.QueueState{}, err + } + store := stores.GetQueueStore() + for { + if err := ctx.Err(); err != nil { + return entity.QueueState{}, err + } + queue, err := store.Get(ctx, name) + missing := storage.IsNotFound(err) + if err != nil && !missing { + return entity.QueueState{}, err + } + if !missing { + if queue.Name != name || queue.Version <= 0 { + return entity.QueueState{}, fmt.Errorf("initialize queue %q: invalid stored queue %q at row version %d: %w", name, queue.Name, queue.Version, queuelog.ErrInconsistentHistory) + } + if queue.EnablementVersion != 0 || queue.Enablement != entity.QueueEnablementUnknown || queue.EffectiveFromCommitURI != "" || queue.EnablementChangedAtMs != 0 { + return c.repairQueueReadModels(ctx, stores, queue) + } + } else { + queue = entity.Queue{Name: name, Version: 1} + } + updated := queue + updated.Enablement, updated.EnablementVersion = entity.QueueEnablementDisabled, 1 + updated.EnablementChangedAtMs = c.now().UnixMilli() + if err := updated.EnablementState().Validate(); err != nil { + return entity.QueueState{}, fmt.Errorf("initialize queue %q: invalid enablement: %w: %v", name, ErrInvalidChange, err) + } + if missing { + if err := store.Create(ctx, updated); errors.Is(err, storage.ErrAlreadyExists) { + continue + } else if err != nil { + return entity.QueueState{}, err + } + } else { + if queue.Version == math.MaxInt32 { + return entity.QueueState{}, fmt.Errorf("initialize queue %q: row version %d exhausted: %w", name, queue.Version, ErrInvalidChange) + } + newVersion := queue.Version + 1 + if err := store.Update(ctx, updated, queue.Version, newVersion); errors.Is(err, storage.ErrVersionMismatch) { + continue + } else if err != nil { + return entity.QueueState{}, err + } + updated.Version = newVersion + } + return c.repairQueueReadModels(ctx, stores, updated) + } +} + +func (c *transitionController) Prepare(ctx context.Context, queue string, enablement entity.QueueEnablement) (PreparedChange, error) { + if enablement != entity.QueueEnablementEnabled && enablement != entity.QueueEnablementDisabled { + return PreparedChange{}, fmt.Errorf("prepare queue %q: unsupported enablement %q: %w", queue, enablement, ErrInvalidChange) + } + stores, err := c.resolveStores(ctx, queue) + if err != nil { + return PreparedChange{}, err + } + internal, err := c.loadInternalQueue(ctx, stores, queue) + if err != nil { + return PreparedChange{}, err + } + current := internal.EnablementState() + change := PreparedChange{previous: current, next: current} + if current.State == enablement { + return change, nil + } + if current.Version == math.MaxInt64 { + return PreparedChange{}, fmt.Errorf("prepare queue %q: enablement version %d exhausted: %w", queue, current.Version, ErrInvalidChange) + } + sc, err := c.sourceControl.For(sourcecontrol.Config{QueueName: queue}) + if err != nil { + return PreparedChange{}, err + } + boundary, err := sc.Latest(ctx) + if err != nil { + return PreparedChange{}, err + } + if boundary == "" || len(boundary) > 255 { + return PreparedChange{}, fmt.Errorf("prepare queue %q: invalid boundary URI %q (length %d): %w", queue, boundary, len(boundary), ErrInvalidChange) + } + if current.EffectiveFromCommitURI != "" { + forward, err := sc.IsAncestor(ctx, current.EffectiveFromCommitURI, boundary) + if err != nil { + return PreparedChange{}, err + } + if !forward { + return PreparedChange{}, fmt.Errorf("prepare queue %q: boundary %q does not include previous boundary %q: %w", queue, boundary, current.EffectiveFromCommitURI, ErrInvalidChange) + } + } + change.next.State, change.next.EffectiveFromCommitURI = enablement, boundary + change.next.ChangedAtMs = c.now().UnixMilli() + if change.next.ChangedAtMs <= 0 { + return PreparedChange{}, fmt.Errorf("prepare queue %q: invalid enablement timestamp %d: %w", queue, change.next.ChangedAtMs, ErrInvalidChange) + } + return change, nil +} + +func (c *transitionController) Apply(ctx context.Context, change PreparedChange) (entity.QueueState, error) { + if err := change.previous.Validate(); err != nil { + return entity.QueueState{}, fmt.Errorf("apply queue %q: invalid prepared state version %d: %w: %v", change.previous.Queue, change.previous.Version, ErrInvalidChange, err) + } + stores, err := c.resolveStores(ctx, change.previous.Queue) + if err != nil { + return entity.QueueState{}, err + } + for { + internal, err := c.loadInternalQueue(ctx, stores, change.previous.Queue) + if err != nil { + return entity.QueueState{}, err + } + current := internal.EnablementState() + if change.next == change.previous { + if current == change.previous { + return current, nil + } + retained, err := queuelog.LoadVersion(ctx, stores.GetQueueLogStore(), current.Queue, change.previous.Version) + if err != nil { + return entity.QueueState{}, err + } + if retained != change.previous { + return entity.QueueState{}, storage.ErrVersionMismatch + } + return retained, nil + } + if change.previous.Version == math.MaxInt64 || internal.Version == math.MaxInt32 { + return entity.QueueState{}, fmt.Errorf("apply queue %q: cannot advance enablement version %d or row version %d: %w", current.Queue, change.previous.Version, internal.Version, ErrInvalidChange) + } + stateVersion := change.previous.Version + 1 + if current.Version != change.previous.Version { + return reconcileChange(ctx, stores.GetQueueLogStore(), current, change, stateVersion) + } + if current != change.previous { + return entity.QueueState{}, storage.ErrVersionMismatch + } + if err := c.validatePreparedBoundary(ctx, change); err != nil { + return entity.QueueState{}, err + } + // An enablement change must preserve concurrent pipeline updates. + updated := internal + updated.Enablement, updated.EnablementVersion = change.next.State, stateVersion + updated.EffectiveFromCommitURI, updated.EnablementChangedAtMs = change.next.EffectiveFromCommitURI, change.next.ChangedAtMs + newVersion := internal.Version + 1 + if err := stores.GetQueueStore().Update(ctx, updated, internal.Version, newVersion); err != nil { + observed, readErr := c.loadInternalQueue(ctx, stores, current.Queue) + if readErr != nil { + return entity.QueueState{}, errors.Join(err, readErr) + } + if observed.EnablementState() == change.previous && errors.Is(err, storage.ErrVersionMismatch) { + continue + } + applied, reconcileErr := reconcileChange(ctx, stores.GetQueueLogStore(), observed.EnablementState(), change, stateVersion) + if reconcileErr != nil { + if errors.Is(reconcileErr, storage.ErrVersionMismatch) { + return entity.QueueState{}, err + } + return entity.QueueState{}, reconcileErr + } + return applied, nil + } + updated.Version = newVersion + return c.repairQueueReadModels(ctx, stores, updated) + } +} + +func (c *transitionController) resolveStores(ctx context.Context, queue string) (storage.Storage, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + if queue == "" || len(queue) > 255 { + return nil, fmt.Errorf("invalid queue name %q (length %d): %w", queue, len(queue), ErrInvalidChange) + } + stores, err := c.stores.For(storage.Config{QueueName: queue}) + if err != nil { + return nil, fmt.Errorf("resolve queue state storage: %w", err) + } + return stores, nil +} + +func (c *transitionController) loadInternalQueue(ctx context.Context, stores storage.Storage, name string) (entity.Queue, error) { + if err := ctx.Err(); err != nil { + return entity.Queue{}, err + } + queue, err := stores.GetQueueStore().Get(ctx, name) + if err != nil { + return entity.Queue{}, err + } + if queue.Name != name || queue.Version <= 0 { + return entity.Queue{}, fmt.Errorf("load queue %q: invalid stored queue %q at row version %d: %w", name, queue.Name, queue.Version, queuelog.ErrInconsistentHistory) + } + if queue.EnablementVersion == 0 && queue.Enablement == entity.QueueEnablementUnknown && queue.EffectiveFromCommitURI == "" && queue.EnablementChangedAtMs == 0 { + return entity.Queue{}, storage.ErrNotFound + } + if _, err := c.repairQueueReadModels(ctx, stores, queue); err != nil { + return entity.Queue{}, err + } + return queue, nil +} + +func (c *transitionController) repairQueueReadModels(ctx context.Context, stores storage.Storage, queue entity.Queue) (entity.QueueState, error) { + state := queue.EnablementState() + if err := state.Validate(); err != nil { + return entity.QueueState{}, fmt.Errorf("queue %q: invalid internal enablement at state version %d, row version %d: %w: %v", queue.Name, state.Version, queue.Version, queuelog.ErrInconsistentHistory, err) + } + return state, c.materializer.PersistLog(ctx, stores, queuelog.NewQueueStateLog(queue)) +} + +func (c *transitionController) validatePreparedBoundary(ctx context.Context, change PreparedChange) error { + sc, err := c.sourceControl.For(sourcecontrol.Config{QueueName: change.previous.Queue}) + if err != nil { + return err + } + tip, err := sc.Latest(ctx) + if err != nil { + return err + } + belongs, err := sc.IsAncestor(ctx, change.next.EffectiveFromCommitURI, tip) + if err != nil { + return err + } + if !belongs { + return fmt.Errorf("queue %q: prepared boundary %q is not an ancestor of current tip %q: %w", change.previous.Queue, change.next.EffectiveFromCommitURI, tip, ErrInvalidChange) + } + if change.previous.EffectiveFromCommitURI != "" { + forward, err := sc.IsAncestor(ctx, change.previous.EffectiveFromCommitURI, change.next.EffectiveFromCommitURI) + if err != nil { + return err + } + if !forward { + return fmt.Errorf("queue %q: prepared boundary %q does not include previous boundary %q: %w", change.previous.Queue, change.next.EffectiveFromCommitURI, change.previous.EffectiveFromCommitURI, ErrInvalidChange) + } + } + return nil +} + +func reconcileChange(ctx context.Context, logs storage.QueueLogStore, current entity.QueueState, change PreparedChange, version int64) (entity.QueueState, error) { + if current.Version < version { + return entity.QueueState{}, storage.ErrVersionMismatch + } + applied := current + if current.Version > version { + var err error + applied, err = queuelog.LoadVersion(ctx, logs, current.Queue, version) + if err != nil { + return entity.QueueState{}, err + } + } + if applied.Queue != change.next.Queue || applied.State != change.next.State || applied.EffectiveFromCommitURI != change.next.EffectiveFromCommitURI || applied.ChangedAtMs != change.next.ChangedAtMs { + return entity.QueueState{}, storage.ErrVersionMismatch + } + return applied, nil +} diff --git a/stovepipe/core/queuestate/queuestate_test.go b/stovepipe/core/queuestate/queuestate_test.go new file mode 100644 index 000000000..fe44dcf15 --- /dev/null +++ b/stovepipe/core/queuestate/queuestate_test.go @@ -0,0 +1,501 @@ +// 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 queuestate + +import ( + "context" + "errors" + "math" + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/stovepipe/core/queuelog" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" + scfake "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/fake" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +type stateFixture struct { + manager *transitionController + state entity.QueueState + queue entity.Queue + summary entity.QueueSummary + summaryCreateErr, summaryUpdateErr error + logs map[string]entity.QueueLog + source *stateSourceFactory + clock int64 + getErr, createErr, updateErr, logCreateErr, logGetErr error + commitOnError bool + beforeUpdate func() +} +type stateStorage struct { + storage.Storage + f *stateFixture +} + +func (s stateStorage) GetQueueStore() storage.QueueStore { return memoryStateStore{s.f} } +func (s stateStorage) GetQueueSummaryStore() storage.QueueSummaryStore { + return memorySummaryStore{s.f} +} +func (s stateStorage) GetQueueLogStore() storage.QueueLogStore { + return memoryStateLogStore{s.f} +} + +type stateStorageFactory struct{ stores storage.Storage } + +func (f stateStorageFactory) For(storage.Config) (storage.Storage, error) { return f.stores, nil } + +type stateSourceFactory struct { + sc sourcecontrol.SourceControl + calls int + err error +} + +func (f *stateSourceFactory) For(sourcecontrol.Config) (sourcecontrol.SourceControl, error) { + f.calls++ + return f.sc, f.err +} + +type fixedLatestSource struct { + sourcecontrol.SourceControl + tip string +} + +func (s fixedLatestSource) Latest(context.Context) (string, error) { return s.tip, nil } + +type memoryStateStore struct{ f *stateFixture } + +func (s memoryStateStore) Get(_ context.Context, name string) (entity.Queue, error) { + if s.f.getErr != nil { + return entity.Queue{}, s.f.getErr + } + if s.f.queue.Version == 0 { + return entity.Queue{}, storage.ErrNotFound + } + queue := s.f.queue + if s.f.state.Queue != "" { + queue.Name = s.f.state.Queue + } + queue.Enablement, queue.EnablementVersion = s.f.state.State, s.f.state.Version + queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs = s.f.state.EffectiveFromCommitURI, s.f.state.ChangedAtMs + return queue, nil +} +func (s memoryStateStore) Create(_ context.Context, queue entity.Queue) error { + if s.f.createErr != nil { + return s.f.createErr + } + if s.f.queue.Version != 0 { + return storage.ErrAlreadyExists + } + s.f.queue, s.f.state = queue, queue.EnablementState() + return nil +} +func (s memoryStateStore) Update(_ context.Context, queue entity.Queue, oldVersion, newVersion int32) error { + if hook := s.f.beforeUpdate; hook != nil { + s.f.beforeUpdate = nil + hook() + } + if s.f.queue.Version != oldVersion { + return storage.ErrVersionMismatch + } + if s.f.updateErr != nil && !s.f.commitOnError { + return s.f.updateErr + } + queue.Version = newVersion + s.f.queue, s.f.state = queue, queue.EnablementState() + return s.f.updateErr +} + +type memorySummaryStore struct{ f *stateFixture } + +func (s memorySummaryStore) Get(context.Context) (entity.QueueSummary, error) { + if s.f.summary.Version == 0 { + return entity.QueueSummary{}, storage.ErrNotFound + } + return s.f.summary, nil +} +func (s memorySummaryStore) Create(_ context.Context, summary entity.QueueSummary) error { + if s.f.summaryCreateErr != nil { + return s.f.summaryCreateErr + } + if s.f.summary.Version != 0 { + return storage.ErrAlreadyExists + } + s.f.summary = summary + return nil +} +func (s memorySummaryStore) Update(_ context.Context, summary entity.QueueSummary, oldVersion, newVersion int64) error { + if s.f.summaryUpdateErr != nil { + return s.f.summaryUpdateErr + } + if s.f.summary.Version != oldVersion { + return storage.ErrVersionMismatch + } + summary.Version = newVersion + s.f.summary = summary + return nil +} + +type memoryStateLogStore struct{ f *stateFixture } + +func (s memoryStateLogStore) Get(_ context.Context, id string) (entity.QueueLog, error) { + if s.f.logGetErr != nil { + return entity.QueueLog{}, s.f.logGetErr + } + log, ok := s.f.logs[id] + if !ok { + return entity.QueueLog{}, storage.ErrNotFound + } + return log, nil +} +func (s memoryStateLogStore) Create(_ context.Context, log entity.QueueLog) error { + if s.f.logCreateErr != nil { + return s.f.logCreateErr + } + if _, ok := s.f.logs[log.ID]; ok { + return storage.ErrAlreadyExists + } + s.f.logs[log.ID] = log + return nil +} +func newStateFixture() *stateFixture { + f := &stateFixture{logs: make(map[string]entity.QueueLog), source: &stateSourceFactory{}, clock: 1000} + f.manager = NewTransitionController(stateStorageFactory{stateStorage{f: f}}, f.source, queuelog.NewMaterializer(tally.NoopScope)).(*transitionController) + f.manager.now = func() time.Time { return time.UnixMilli(f.clock) } + f.setTip("A") + return f +} +func (f *stateFixture) setTip(tip string) { + history := []string{"G", "F", "E", "D", "C", "B", "A", "before"} + for i, uri := range history { + if uri == tip { + f.source.sc = scfake.New(sourcecontrol.Config{QueueName: "repo/main"}, history[i:]) + return + } + } + f.source.sc = scfake.New(sourcecontrol.Config{QueueName: "repo/main"}, []string{tip}) +} +func (f *stateFixture) applyState(t *testing.T, state entity.QueueEnablement, tip string) (PreparedChange, entity.QueueState) { + t.Helper() + f.setTip(tip) + change, err := f.manager.Prepare(context.Background(), "repo/main", state) + require.NoError(t, err) + applied, err := f.manager.Apply(context.Background(), change) + require.NoError(t, err) + return change, applied +} +func initializeFixture(t *testing.T, f *stateFixture) entity.QueueState { + t.Helper() + state, err := f.manager.Initialize(context.Background(), "repo/main") + require.NoError(t, err) + return state +} + +func TestStateInitializationPreservesStateAndRepairsHistory(t *testing.T) { + for _, partial := range []bool{false, true} { + t.Run(map[bool]string{false: "complete", true: "history write interrupted"}[partial], func(t *testing.T) { + f := newStateFixture() + if partial { + f.logCreateErr = errors.New("history unavailable") + _, err := f.manager.Initialize(context.Background(), "repo/main") + require.ErrorIs(t, err, f.logCreateErr) + require.Equal(t, int64(1), f.state.Version) + require.Empty(t, f.logs) + f.logCreateErr = nil + f.clock = 9000 + } + initial := initializeFixture(t, f) + require.Equal(t, int64(1000), initial.ChangedAtMs) + require.Len(t, f.logs, 1) + _, enabled := f.applyState(t, entity.QueueEnablementEnabled, "A") + require.Equal(t, enabled, initializeFixture(t, f)) + }) + } +} + +func TestStateCommittedHistoryAndPreparedReplay(t *testing.T) { + f := newStateFixture() + initializeFixture(t, f) + first, enabled := f.applyState(t, entity.QueueEnablementEnabled, "A") + _, disabled := f.applyState(t, entity.QueueEnablementDisabled, "D") + _, reenabled := f.applyState(t, entity.QueueEnablementEnabled, "G") + require.Equal(t, int64(2), enabled.Version) + require.Equal(t, int64(3), disabled.Version) + require.Equal(t, int64(4), reenabled.Version) + replay, err := f.manager.Apply(context.Background(), first) + require.NoError(t, err) + require.Equal(t, enabled, replay) + require.Equal(t, reenabled, f.state) + require.Len(t, f.logs, 4) +} + +func TestStateNoOpPreservesBoundaryAndReplays(t *testing.T) { + f := newStateFixture() + initial := initializeFixture(t, f) + noop, err := f.manager.Prepare(context.Background(), "repo/main", entity.QueueEnablementDisabled) + require.NoError(t, err) + require.Zero(t, f.source.calls) + state, err := f.manager.Apply(context.Background(), noop) + require.NoError(t, err) + require.Equal(t, initial, state) + f.applyState(t, entity.QueueEnablementEnabled, "A") + replay, err := f.manager.Apply(context.Background(), noop) + require.NoError(t, err) + require.Equal(t, initial, replay) + require.Len(t, f.logs, 2) +} + +func TestStatePartialLogWriteBlocksNextTransition(t *testing.T) { + f := newStateFixture() + initializeFixture(t, f) + change, err := f.manager.Prepare(context.Background(), "repo/main", entity.QueueEnablementEnabled) + require.NoError(t, err) + f.beforeUpdate = func() { f.logCreateErr = errors.New("history unavailable") } + _, err = f.manager.Apply(context.Background(), change) + require.ErrorIs(t, err, f.logCreateErr) + require.Equal(t, int64(2), f.state.Version) + require.Len(t, f.logs, 1) + _, err = f.manager.Prepare(context.Background(), "repo/main", entity.QueueEnablementDisabled) + require.ErrorIs(t, err, f.logCreateErr) + require.Equal(t, int64(2), f.state.Version) + f.logCreateErr = nil + f.clock = 9000 + noop, err := f.manager.Prepare(context.Background(), "repo/main", entity.QueueEnablementEnabled) + require.NoError(t, err) + _, err = f.manager.Apply(context.Background(), noop) + require.NoError(t, err) + replay, err := f.manager.Apply(context.Background(), change) + require.NoError(t, err) + require.Equal(t, int64(1000), replay.ChangedAtMs) + require.Len(t, f.logs, 2) + f.applyState(t, entity.QueueEnablementDisabled, "D") + require.Len(t, f.logs, 3) +} + +func TestStateUncertainWriteReconciliation(t *testing.T) { + for _, committed := range []bool{false, true} { + t.Run(map[bool]string{false: "failed before commit", true: "committed before connection failed"}[committed], func(t *testing.T) { + f := newStateFixture() + initializeFixture(t, f) + change, err := f.manager.Prepare(context.Background(), "repo/main", entity.QueueEnablementEnabled) + require.NoError(t, err) + f.updateErr = errors.New("connection lost") + f.commitOnError = committed + state, err := f.manager.Apply(context.Background(), change) + if committed { + require.NoError(t, err) + require.Equal(t, int64(2), state.Version) + } else { + require.ErrorIs(t, err, f.updateErr) + require.Equal(t, int64(1), f.state.Version) + require.Len(t, f.logs, 1) + } + f.updateErr = nil + f.clock = 9999 + f.setTip("D") + state, err = f.manager.Apply(context.Background(), change) + require.NoError(t, err) + require.Equal(t, "A", state.EffectiveFromCommitURI) + require.Equal(t, int64(1000), state.ChangedAtMs) + require.Len(t, f.logs, 2) + }) + } +} + +func TestStateCompetingPreparedChanges(t *testing.T) { + f := newStateFixture() + initializeFixture(t, f) + first, err := f.manager.Prepare(context.Background(), "repo/main", entity.QueueEnablementEnabled) + require.NoError(t, err) + f.setTip("B") + second, err := f.manager.Prepare(context.Background(), "repo/main", entity.QueueEnablementEnabled) + require.NoError(t, err) + f.beforeUpdate = func() { _, err := f.manager.Apply(context.Background(), second); require.NoError(t, err) } + _, err = f.manager.Apply(context.Background(), first) + require.ErrorIs(t, err, storage.ErrVersionMismatch) + require.Equal(t, "B", f.state.EffectiveFromCommitURI) + require.Len(t, f.logs, 2) + _, err = f.manager.Apply(context.Background(), first) + require.ErrorIs(t, err, storage.ErrVersionMismatch) +} + +func TestStateBoundaryAndDecisionValidation(t *testing.T) { + for _, tc := range []struct { + name string + tip string + state entity.QueueEnablement + want error + }{ + {"equal boundary", "A", entity.QueueEnablementDisabled, nil}, + {"forward boundary", "D", entity.QueueEnablementDisabled, nil}, + {"backward boundary", "before", entity.QueueEnablementDisabled, ErrInvalidChange}, + {"rewritten lineage", "other", entity.QueueEnablementDisabled, sourcecontrol.ErrNotFound}, + {"unknown enablement", "D", entity.QueueEnablementUnknown, ErrInvalidChange}, + } { + t.Run(tc.name, func(t *testing.T) { + f := newStateFixture() + initializeFixture(t, f) + f.applyState(t, entity.QueueEnablementEnabled, "A") + f.setTip(tc.tip) + if tc.tip == "before" { + f.source.sc = fixedLatestSource{SourceControl: scfake.New(sourcecontrol.Config{QueueName: "repo/main"}, []string{"G", "A", "before"}), tip: "before"} + } + change, err := f.manager.Prepare(context.Background(), "repo/main", tc.state) + if tc.want != nil { + require.ErrorIs(t, err, tc.want) + require.Equal(t, int64(2), f.state.Version) + return + } + require.NoError(t, err) + _, err = f.manager.Apply(context.Background(), change) + require.NoError(t, err) + require.Equal(t, int64(3), f.state.Version) + }) + } + f := newStateFixture() + initializeFixture(t, f) + change, err := f.manager.Prepare(context.Background(), "repo/main", entity.QueueEnablementEnabled) + require.NoError(t, err) + f.setTip("other") + _, err = f.manager.Apply(context.Background(), change) + require.ErrorIs(t, err, sourcecontrol.ErrNotFound) + require.Equal(t, int64(1), f.state.Version) +} + +func TestStateFailuresAndHistoryCorruption(t *testing.T) { + for _, tc := range []struct { + name string + corrupt func(*stateFixture) + }{ + {"conflicting current log", func(f *stateFixture) { + log := f.logs[entity.QueueLogID(1)] + log.TimestampMs++ + f.logs[log.ID] = log + }}, + {"wrong queue", func(f *stateFixture) { f.state.Queue = "other" }}, + {"invalid initial state", func(f *stateFixture) { f.state.State = entity.QueueEnablementEnabled }}, + {"invalid version", func(f *stateFixture) { f.state.Version = -1 }}, + } { + t.Run(tc.name, func(t *testing.T) { + f := newStateFixture() + initializeFixture(t, f) + tc.corrupt(f) + _, err := f.manager.Initialize(context.Background(), "repo/main") + require.ErrorIs(t, err, queuelog.ErrInconsistentHistory) + }) + } + f := newStateFixture() + initializeFixture(t, f) + _, err := queuelog.LoadVersion(context.Background(), memoryStateLogStore{f}, "repo/main", 2) + require.ErrorIs(t, err, queuelog.ErrInconsistentHistory) + require.False(t, storage.IsNotFound(err)) + f.state.Version = math.MaxInt64 + f.state.EffectiveFromCommitURI = "A" + _, err = f.manager.Prepare(context.Background(), "repo/main", entity.QueueEnablementEnabled) + require.ErrorIs(t, err, ErrInvalidChange) + _, err = f.manager.Apply(context.Background(), PreparedChange{}) + require.ErrorIs(t, err, ErrInvalidChange) + _, err = f.manager.Initialize(context.Background(), "") + require.ErrorIs(t, err, ErrInvalidChange) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err = f.manager.Initialize(ctx, "repo/main") + require.ErrorIs(t, err, context.Canceled) +} + +func TestStateDependencyFailuresRemainVisible(t *testing.T) { + failure := errors.New("dependency unavailable") + f := newStateFixture() + f.createErr = failure + _, err := f.manager.Initialize(context.Background(), "repo/main") + require.ErrorIs(t, err, failure) + f.createErr = nil + initializeFixture(t, f) + f.getErr = failure + _, err = f.manager.Prepare(context.Background(), "repo/main", entity.QueueEnablementEnabled) + require.ErrorIs(t, err, failure) + f.getErr = nil + f.source.err = failure + _, err = f.manager.Prepare(context.Background(), "repo/main", entity.QueueEnablementEnabled) + require.ErrorIs(t, err, failure) + f.source.err = nil + f.logGetErr = failure + _, err = f.manager.Initialize(context.Background(), "repo/main") + require.ErrorIs(t, err, failure) +} + +func TestStateChangesPreserveOperationalQueueUpdates(t *testing.T) { + for _, duringWrite := range []bool{false, true} { + t.Run(map[bool]string{false: "between prepare and apply", true: "CAS conflict during apply"}[duringWrite], func(t *testing.T) { + f := newStateFixture() + initializeFixture(t, f) + change, err := f.manager.Prepare(context.Background(), "repo/main", entity.QueueEnablementEnabled) + require.NoError(t, err) + mutateQueue := func() { + f.queue.Version++ + f.queue.LastGreenURI = "green" + f.queue.InFlightCount = 3 + f.queue.LatestRequestID = "7" + } + if duringWrite { + f.beforeUpdate = mutateQueue + } else { + mutateQueue() + } + applied, err := f.manager.Apply(context.Background(), change) + require.NoError(t, err) + require.Equal(t, int64(2), applied.Version) + require.Equal(t, int32(3), f.queue.Version) + require.Equal(t, "green", f.queue.LastGreenURI) + require.Equal(t, int32(3), f.queue.InFlightCount) + require.Equal(t, "7", f.queue.LatestRequestID) + require.Equal(t, applied, f.summary.Snapshot()) + require.Len(t, f.logs, 2) + }) + } +} + +func TestStateInitializationAdoptsExistingOperationalQueue(t *testing.T) { + f := newStateFixture() + f.queue = entity.Queue{Name: "repo/main", Version: 12, InFlightCount: 2, LastGreenURI: "green", LatestRequestID: "9"} + initial := initializeFixture(t, f) + require.Equal(t, int64(1), initial.Version) + require.Equal(t, int32(13), f.queue.Version) + require.Equal(t, int32(2), f.queue.InFlightCount) + require.Equal(t, "green", f.queue.LastGreenURI) + require.Equal(t, "9", f.queue.LatestRequestID) + require.Equal(t, initial, f.summary.Snapshot()) +} + +func TestStateProjectionFailureDoesNotLoseAppliedState(t *testing.T) { + f := newStateFixture() + initial := initializeFixture(t, f) + change, err := f.manager.Prepare(context.Background(), "repo/main", entity.QueueEnablementEnabled) + require.NoError(t, err) + f.beforeUpdate = func() { f.summaryUpdateErr = errors.New("projection unavailable") } + _, err = f.manager.Apply(context.Background(), change) + require.ErrorIs(t, err, f.summaryUpdateErr) + require.Equal(t, int64(2), f.state.Version) + require.Equal(t, initial, f.summary.Snapshot()) + require.Len(t, f.logs, 2) + f.summaryUpdateErr = nil + replay, err := f.manager.Apply(context.Background(), change) + require.NoError(t, err) + require.Equal(t, f.state, replay) + require.Equal(t, replay, f.summary.Snapshot()) +} diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index b38416e00..e2a9c140c 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -9,6 +9,9 @@ go_library( "project_status.go", "queue.go", "queue_config.go", + "queue_log.go", + "queue_state.go", + "queue_summary.go", "request.go", "request_acceptance.go", "request_history.go", @@ -26,6 +29,7 @@ go_test( name = "go_default_test", srcs = [ "build_test.go", + "queue_state_test.go", "request_id_test.go", "request_log_test.go", "request_test.go", diff --git a/stovepipe/entity/queue.go b/stovepipe/entity/queue.go index 23d31efba..9ca1c8fcc 100644 --- a/stovepipe/entity/queue.go +++ b/stovepipe/entity/queue.go @@ -43,7 +43,25 @@ type Queue struct { // Empty until the first request is created. Coalescing compares IDs via CompareRequestID. LatestRequestID string `json:"latest_request_id"` + // Enablement is the applied enabled/disabled state; empty means unregistered. + Enablement QueueEnablement `json:"enablement"` + + // EnablementVersion is the queue-scoped enablement sequence; zero means unregistered. + // Operational changes do not advance this sequence. + EnablementVersion int64 `json:"enablement_version"` + + // EffectiveFromCommitURI is the inclusive immutable enablement boundary; empty for initial registration. + EffectiveFromCommitURI string `json:"effective_from_commit_uri"` + + // EnablementChangedAtMs is the enablement application timestamp in Unix milliseconds. + EnablementChangedAtMs int64 `json:"enablement_changed_at_ms"` + // Version is the version of the object. It is used for optimistic locking. // Versioning starts at 1 and is incremented for each change to the object. Version int32 `json:"version"` } + +func (q Queue) EnablementState() QueueState { + return QueueState{Queue: q.Name, State: q.Enablement, Version: q.EnablementVersion, + EffectiveFromCommitURI: q.EffectiveFromCommitURI, ChangedAtMs: q.EnablementChangedAtMs} +} diff --git a/stovepipe/entity/queue_log.go b/stovepipe/entity/queue_log.go new file mode 100644 index 000000000..2384fea7c --- /dev/null +++ b/stovepipe/entity/queue_log.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 entity + +import "fmt" + +// QueueLog is one immutable occurrence of a committed enablement state. +type QueueLog struct { + // ID is the stable queue-scoped occurrence identity derived from QueueStateVersion. + ID string + // Queue is the stable repo-and-ref namespace. + Queue string + // TimestampMs is the application's timestamp in Unix milliseconds. + TimestampMs int64 + // State is the committed enabled or disabled value. + State QueueEnablement + // QueueStateVersion is the positive version identifying the applied state. + QueueStateVersion int64 + // EffectiveFromCommitURI is the inclusive immutable boundary; empty only at version 1. + EffectiveFromCommitURI string +} + +// QueueLogID returns the stable occurrence identity for a state version. +func QueueLogID(version int64) string { + return fmt.Sprintf("state/%d", version) +} + +func (l QueueLog) Snapshot() QueueState { + return QueueState{Queue: l.Queue, State: l.State, Version: l.QueueStateVersion, + EffectiveFromCommitURI: l.EffectiveFromCommitURI, ChangedAtMs: l.TimestampMs} +} + +func (l QueueLog) Validate() error { + if l.ID != QueueLogID(l.QueueStateVersion) { + return fmt.Errorf("queue state log identity differs from its state version") + } + return l.Snapshot().Validate() +} diff --git a/stovepipe/entity/queue_state.go b/stovepipe/entity/queue_state.go new file mode 100644 index 000000000..2ee1f1e26 --- /dev/null +++ b/stovepipe/entity/queue_state.go @@ -0,0 +1,58 @@ +// 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 + +import "fmt" + +// QueueEnablement is applied enablement, independent of execution controls and verdicts. +type QueueEnablement string + +const ( + // QueueEnablementUnknown is unset and never an explicit disablement. + QueueEnablementUnknown QueueEnablement = "" + QueueEnablementEnabled QueueEnablement = "enabled" + QueueEnablementDisabled QueueEnablement = "disabled" +) + +// QueueState describes applied enablement, independent of storage row versions. +type QueueState struct { + // Queue is the stable repo-and-ref namespace, at most 255 bytes. + Queue string + // State is enabled or disabled. + State QueueEnablement + // EffectiveFromCommitURI is the inclusive immutable boundary, at most 255 bytes; empty only at version 1. + EffectiveFromCommitURI string + // ChangedAtMs is the application timestamp in Unix milliseconds. + ChangedAtMs int64 + // Version is the positive queue-scoped enablement sequence. + Version int64 +} + +func (s QueueState) Validate() error { + if s.Queue == "" || len(s.Queue) > 255 || s.Version <= 0 || s.ChangedAtMs <= 0 { + return fmt.Errorf("invalid queue state identity, version or timestamp") + } + if s.State != QueueEnablementEnabled && s.State != QueueEnablementDisabled { + return fmt.Errorf("invalid queue enablement %q", s.State) + } + if s.Version == 1 { + if s.State != QueueEnablementDisabled || s.EffectiveFromCommitURI != "" { + return fmt.Errorf("initial queue state must be disabled without a boundary") + } + } else if s.EffectiveFromCommitURI == "" || len(s.EffectiveFromCommitURI) > 255 { + return fmt.Errorf("invalid queue state commit boundary") + } + return nil +} diff --git a/stovepipe/entity/queue_state_test.go b/stovepipe/entity/queue_state_test.go new file mode 100644 index 000000000..c4d2cbd46 --- /dev/null +++ b/stovepipe/entity/queue_state_test.go @@ -0,0 +1,58 @@ +// 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 + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestQueueStateInvariants(t *testing.T) { + initial := QueueState{Queue: "repo/main", State: QueueEnablementDisabled, ChangedAtMs: 1000, Version: 1} + for _, tc := range []struct { + name string + change func(*QueueState) + valid bool + }{ + {"initial", func(*QueueState) {}, true}, + {"enabled occurrence", func(s *QueueState) { s.Version = 2; s.State = QueueEnablementEnabled; s.EffectiveFromCommitURI = "A" }, true}, + {"missing queue", func(s *QueueState) { s.Queue = "" }, false}, + {"oversized queue", func(s *QueueState) { s.Queue = strings.Repeat("x", 256) }, false}, + {"unknown state", func(s *QueueState) { s.State = QueueEnablementUnknown }, false}, + {"invalid version", func(s *QueueState) { s.Version = 0 }, false}, + {"missing timestamp", func(s *QueueState) { s.ChangedAtMs = 0 }, false}, + {"enabled initial", func(s *QueueState) { s.State = QueueEnablementEnabled }, false}, + {"initial boundary", func(s *QueueState) { s.EffectiveFromCommitURI = "A" }, false}, + {"missing later boundary", func(s *QueueState) { s.Version = 2 }, false}, + {"oversized boundary", func(s *QueueState) { s.Version = 2; s.EffectiveFromCommitURI = strings.Repeat("x", 256) }, false}, + } { + t.Run(tc.name, func(t *testing.T) { + state := initial + tc.change(&state) + if tc.valid { + require.NoError(t, state.Validate()) + } else { + require.Error(t, state.Validate()) + } + }) + } + log := QueueLog{ID: QueueLogID(1), Queue: initial.Queue, TimestampMs: initial.ChangedAtMs, State: initial.State, QueueStateVersion: 1} + require.NoError(t, log.Validate()) + require.Equal(t, initial, log.Snapshot()) + log.ID = "other" + require.Error(t, log.Validate()) +} diff --git a/stovepipe/entity/queue_summary.go b/stovepipe/entity/queue_summary.go new file mode 100644 index 000000000..ba2da65ee --- /dev/null +++ b/stovepipe/entity/queue_summary.go @@ -0,0 +1,45 @@ +// 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 + +import "fmt" + +// QueueSummary is the current read model derived from retained queue history. +type QueueSummary struct { + // Queue is the stable repo-and-ref namespace, at most 255 bytes. + Queue string + // State is the latest materialized enabled/disabled value. + State QueueEnablement + // QueueStateVersion identifies the applied enablement occurrence represented here. + QueueStateVersion int64 + // EffectiveFromCommitURI is the inclusive immutable boundary; empty at state version 1. + EffectiveFromCommitURI string + // StateTimestampMs is the occurrence timestamp in Unix milliseconds. + StateTimestampMs int64 + // Version is the positive optimistic-lock version of this projection. + Version int64 +} + +func (s QueueSummary) Snapshot() QueueState { + return QueueState{Queue: s.Queue, State: s.State, Version: s.QueueStateVersion, + EffectiveFromCommitURI: s.EffectiveFromCommitURI, ChangedAtMs: s.StateTimestampMs} +} + +func (s QueueSummary) Validate() error { + if s.Version <= 0 { + return fmt.Errorf("invalid queue summary version") + } + return s.Snapshot().Validate() +} diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index 571ac6eaa..fc12fdf84 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -4,7 +4,9 @@ go_library( name = "go_default_library", srcs = [ "build_store.go", + "queue_log_store.go", "queue_store.go", + "queue_summary_store.go", "request_acceptance_store.go", "request_log_store.go", "request_store.go", diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index 2bf3c9870..de79c8959 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -21,3 +21,5 @@ A `Get` immediately following a successful write (`Create`/`Update`) — by the Same design space as [`submitqueue/extension/storage`](../../../submitqueue/extension/storage/README.md#key-value-contract): every method must be satisfiable by a plain key-value backend as cheaply as by MySQL — get/put/conditional-update by primary key only, no query-by-attribute or server-side filtering. `RequestURIStore` is the reverse-lookup example here (which request owns a given commit URI), kept as its own store rather than a secondary index on `RequestStore`. `RequestAcceptanceStore` is the ordered lookup mapping for List: immutable `(queue, accepted_at_ms, request_id)` keys with bounded primary-key range scans, not a secondary index or a second status projection. The request-log materializer ensures each known acceptance has a mapping after summary persistence, including retries where the summary is unchanged. Unknown acceptance times are not mapped; historical backfill is separate. + +`QueueStore` owns internal enablement alongside pipeline coordination. `QueueSummaryStore` persists the current read model projected from immutable `QueueLogStore` occurrences. Enablement sequences, operational row versions and projection versions are distinct. These stores use primary-key operations; version arithmetic, duplicate reconciliation and projection repair belong to business logic. See [queue state](../../../doc/rfc/stovepipe/queue-state.md). diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index 9db6f3d8a..dab7cef72 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -4,7 +4,9 @@ go_library( name = "go_default_library", srcs = [ "build_store_mock.go", + "queue_log_store_mock.go", "queue_store_mock.go", + "queue_summary_store_mock.go", "request_acceptance_store_mock.go", "request_log_store_mock.go", "request_store_mock.go", diff --git a/stovepipe/extension/storage/mock/queue_log_store_mock.go b/stovepipe/extension/storage/mock/queue_log_store_mock.go new file mode 100644 index 000000000..b6697734c --- /dev/null +++ b/stovepipe/extension/storage/mock/queue_log_store_mock.go @@ -0,0 +1,71 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: queue_log_store.go +// +// Generated by this command: +// +// mockgen -source=queue_log_store.go -destination=mock/queue_log_store_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/stovepipe/entity" + gomock "go.uber.org/mock/gomock" +) + +// MockQueueLogStore is a mock of QueueLogStore interface. +type MockQueueLogStore struct { + ctrl *gomock.Controller + recorder *MockQueueLogStoreMockRecorder + isgomock struct{} +} + +// MockQueueLogStoreMockRecorder is the mock recorder for MockQueueLogStore. +type MockQueueLogStoreMockRecorder struct { + mock *MockQueueLogStore +} + +// NewMockQueueLogStore creates a new mock instance. +func NewMockQueueLogStore(ctrl *gomock.Controller) *MockQueueLogStore { + mock := &MockQueueLogStore{ctrl: ctrl} + mock.recorder = &MockQueueLogStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockQueueLogStore) EXPECT() *MockQueueLogStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockQueueLogStore) Create(arg0 context.Context, arg1 entity.QueueLog) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", arg0, arg1) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockQueueLogStoreMockRecorder) Create(arg0, arg1 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockQueueLogStore)(nil).Create), arg0, arg1) +} + +// Get mocks base method. +func (m *MockQueueLogStore) Get(ctx context.Context, logID string) (entity.QueueLog, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, logID) + ret0, _ := ret[0].(entity.QueueLog) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockQueueLogStoreMockRecorder) Get(ctx, logID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockQueueLogStore)(nil).Get), ctx, logID) +} diff --git a/stovepipe/extension/storage/mock/queue_summary_store_mock.go b/stovepipe/extension/storage/mock/queue_summary_store_mock.go new file mode 100644 index 000000000..80a30c12a --- /dev/null +++ b/stovepipe/extension/storage/mock/queue_summary_store_mock.go @@ -0,0 +1,85 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: queue_summary_store.go +// +// Generated by this command: +// +// mockgen -source=queue_summary_store.go -destination=mock/queue_summary_store_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/stovepipe/entity" + gomock "go.uber.org/mock/gomock" +) + +// MockQueueSummaryStore is a mock of QueueSummaryStore interface. +type MockQueueSummaryStore struct { + ctrl *gomock.Controller + recorder *MockQueueSummaryStoreMockRecorder + isgomock struct{} +} + +// MockQueueSummaryStoreMockRecorder is the mock recorder for MockQueueSummaryStore. +type MockQueueSummaryStoreMockRecorder struct { + mock *MockQueueSummaryStore +} + +// NewMockQueueSummaryStore creates a new mock instance. +func NewMockQueueSummaryStore(ctrl *gomock.Controller) *MockQueueSummaryStore { + mock := &MockQueueSummaryStore{ctrl: ctrl} + mock.recorder = &MockQueueSummaryStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockQueueSummaryStore) EXPECT() *MockQueueSummaryStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockQueueSummaryStore) Create(arg0 context.Context, arg1 entity.QueueSummary) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", arg0, arg1) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockQueueSummaryStoreMockRecorder) Create(arg0, arg1 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockQueueSummaryStore)(nil).Create), arg0, arg1) +} + +// Get mocks base method. +func (m *MockQueueSummaryStore) Get(arg0 context.Context) (entity.QueueSummary, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", arg0) + ret0, _ := ret[0].(entity.QueueSummary) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockQueueSummaryStoreMockRecorder) Get(arg0 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockQueueSummaryStore)(nil).Get), arg0) +} + +// Update mocks base method. +func (m *MockQueueSummaryStore) Update(ctx context.Context, summary entity.QueueSummary, oldVersion, newVersion int64) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Update", ctx, summary, oldVersion, newVersion) + ret0, _ := ret[0].(error) + return ret0 +} + +// Update indicates an expected call of Update. +func (mr *MockQueueSummaryStoreMockRecorder) Update(ctx, summary, oldVersion, newVersion any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Update", reflect.TypeOf((*MockQueueSummaryStore)(nil).Update), ctx, summary, oldVersion, newVersion) +} diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go index 10ca35261..11d137b25 100644 --- a/stovepipe/extension/storage/mock/storage_mock.go +++ b/stovepipe/extension/storage/mock/storage_mock.go @@ -93,6 +93,20 @@ func (mr *MockStorageMockRecorder) GetBuildStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetBuildStore", reflect.TypeOf((*MockStorage)(nil).GetBuildStore)) } +// GetQueueLogStore mocks base method. +func (m *MockStorage) GetQueueLogStore() storage.QueueLogStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetQueueLogStore") + ret0, _ := ret[0].(storage.QueueLogStore) + return ret0 +} + +// GetQueueLogStore indicates an expected call of GetQueueLogStore. +func (mr *MockStorageMockRecorder) GetQueueLogStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetQueueLogStore", reflect.TypeOf((*MockStorage)(nil).GetQueueLogStore)) +} + // GetQueueStore mocks base method. func (m *MockStorage) GetQueueStore() storage.QueueStore { m.ctrl.T.Helper() @@ -107,6 +121,20 @@ func (mr *MockStorageMockRecorder) GetQueueStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetQueueStore", reflect.TypeOf((*MockStorage)(nil).GetQueueStore)) } +// GetQueueSummaryStore mocks base method. +func (m *MockStorage) GetQueueSummaryStore() storage.QueueSummaryStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetQueueSummaryStore") + ret0, _ := ret[0].(storage.QueueSummaryStore) + return ret0 +} + +// GetQueueSummaryStore indicates an expected call of GetQueueSummaryStore. +func (mr *MockStorageMockRecorder) GetQueueSummaryStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetQueueSummaryStore", reflect.TypeOf((*MockStorage)(nil).GetQueueSummaryStore)) +} + // GetRequestAcceptanceStore mocks base method. func (m *MockStorage) GetRequestAcceptanceStore() storage.RequestAcceptanceStore { m.ctrl.T.Helper() diff --git a/stovepipe/extension/storage/mysql/BUILD.bazel b/stovepipe/extension/storage/mysql/BUILD.bazel index a5dd86838..26edd5e8b 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -4,7 +4,9 @@ go_library( name = "go_default_library", srcs = [ "build_store.go", + "queue_log_store.go", "queue_store.go", + "queue_summary_store.go", "request_acceptance_store.go", "request_log_store.go", "request_store.go", @@ -29,6 +31,7 @@ go_test( srcs = [ "build_store_test.go", "queue_store_test.go", + "queue_summary_store_test.go", "request_acceptance_store_test.go", "request_log_store_test.go", "request_store_test.go", diff --git a/stovepipe/extension/storage/mysql/queue_log_store.go b/stovepipe/extension/storage/mysql/queue_log_store.go new file mode 100644 index 000000000..c111a7e0e --- /dev/null +++ b/stovepipe/extension/storage/mysql/queue_log_store.go @@ -0,0 +1,73 @@ +// 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/uber-go/tally" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +type queueLogStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +func NewQueueLogStore(db *sql.DB, scope tally.Scope, queue string) storage.QueueLogStore { + return &queueLogStore{db: db, scope: scope, queue: queue} +} + +func (s *queueLogStore) Get(ctx context.Context, id string) (log entity.QueueLog, retErr error) { + op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + err := s.db.QueryRowContext(ctx, + "SELECT queue, log_id, timestamp_ms, state, queue_state_version, effective_from_commit_uri FROM queue_log WHERE queue = ? AND log_id = ?", s.queue, id). + Scan(&log.Queue, &log.ID, &log.TimestampMs, &log.State, &log.QueueStateVersion, &log.EffectiveFromCommitURI) + if errors.Is(err, sql.ErrNoRows) { + return entity.QueueLog{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.QueueLog{}, fmt.Errorf("read queue state log %q: %w", id, err) + } + return log, nil +} + +func (s *queueLogStore) Create(ctx context.Context, log entity.QueueLog) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + if log.Queue != s.queue { + return fmt.Errorf("log queue differs from bound queue %q", s.queue) + } + if err := log.Validate(); err != nil { + return err + } + _, err := s.db.ExecContext(ctx, + "INSERT INTO queue_log (queue, log_id, timestamp_ms, state, queue_state_version, effective_from_commit_uri) VALUES (?, ?, ?, ?, ?, ?)", + log.Queue, log.ID, log.TimestampMs, log.State, log.QueueStateVersion, log.EffectiveFromCommitURI) + if isDuplicateEntry(err) { + return storage.ErrAlreadyExists + } + if err != nil { + return fmt.Errorf("create queue state log %q: %w", log.ID, err) + } + return nil +} diff --git a/stovepipe/extension/storage/mysql/queue_store.go b/stovepipe/extension/storage/mysql/queue_store.go index aaf3cb169..c4fa43ca3 100644 --- a/stovepipe/extension/storage/mysql/queue_store.go +++ b/stovepipe/extension/storage/mysql/queue_store.go @@ -49,14 +49,15 @@ func (q *queueStore) Create(ctx context.Context, queue entity.Queue) (retErr err } _, err := q.db.ExecContext(ctx, - `INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id) - VALUES (?, ?, ?, ?, ?, ?)`, + `INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version, queue.LastGreenRequestID, + queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs, ) if err != nil { if isDuplicateEntry(err) { @@ -78,7 +79,7 @@ func (q *queueStore) Get(ctx context.Context, name string) (ret entity.Queue, re var queue entity.Queue err := q.db.QueryRowContext(ctx, - "SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id FROM queue WHERE name = ?", + "SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms FROM queue WHERE name = ?", name, ).Scan( &queue.Name, @@ -87,6 +88,7 @@ func (q *queueStore) Get(ctx context.Context, name string) (ret entity.Queue, re &queue.LatestRequestID, &queue.Version, &queue.LastGreenRequestID, + &queue.Enablement, &queue.EnablementVersion, &queue.EffectiveFromCommitURI, &queue.EnablementChangedAtMs, ) if errors.Is(err, sql.ErrNoRows) { @@ -111,13 +113,14 @@ func (q *queueStore) Update(ctx context.Context, queue entity.Queue, oldVersion, result, err := q.db.ExecContext(ctx, `UPDATE queue - SET last_green_uri = ?, in_flight_count = ?, latest_request_id = ?, version = ?, last_green_request_id = ? + SET last_green_uri = ?, in_flight_count = ?, latest_request_id = ?, version = ?, last_green_request_id = ?, enablement = ?, enablement_version = ?, effective_from_commit_uri = ?, enablement_changed_at_ms = ? WHERE name = ? AND version = ?`, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, + queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs, queue.Name, oldVersion, ) diff --git a/stovepipe/extension/storage/mysql/queue_store_test.go b/stovepipe/extension/storage/mysql/queue_store_test.go index 35f5b2749..e96365daf 100644 --- a/stovepipe/extension/storage/mysql/queue_store_test.go +++ b/stovepipe/extension/storage/mysql/queue_store_test.go @@ -60,7 +60,7 @@ func TestQueueStore_Create(t *testing.T) { name: "success", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO queue"). - WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version, queue.LastGreenRequestID). + WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -68,7 +68,7 @@ func TestQueueStore_Create(t *testing.T) { name: "duplicate name returns ErrAlreadyExists", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO queue"). - WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version, queue.LastGreenRequestID). + WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs). WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) }, wantErr: true, @@ -78,7 +78,7 @@ func TestQueueStore_Create(t *testing.T) { name: "other exec error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO queue"). - WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version, queue.LastGreenRequestID). + WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -128,9 +128,9 @@ func TestQueueStore_Get(t *testing.T) { name: "found", queueName: want.Name, setup: func(mock sqlmock.Sqlmock) { - rows := sqlmock.NewRows([]string{"name", "last_green_uri", "in_flight_count", "latest_request_id", "version", "last_green_request_id"}). - AddRow(want.Name, want.LastGreenURI, want.InFlightCount, want.LatestRequestID, want.Version, want.LastGreenRequestID) - mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id FROM queue"). + rows := sqlmock.NewRows([]string{"name", "last_green_uri", "in_flight_count", "latest_request_id", "version", "last_green_request_id", "enablement", "enablement_version", "effective_from_commit_uri", "enablement_changed_at_ms"}). + AddRow(want.Name, want.LastGreenURI, want.InFlightCount, want.LatestRequestID, want.Version, want.LastGreenRequestID, want.Enablement, want.EnablementVersion, want.EffectiveFromCommitURI, want.EnablementChangedAtMs) + mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms FROM queue"). WithArgs(want.Name). WillReturnRows(rows) }, @@ -140,7 +140,7 @@ func TestQueueStore_Get(t *testing.T) { name: "not found", queueName: want.Name, setup: func(mock sqlmock.Sqlmock) { - mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id FROM queue"). + mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms FROM queue"). WithArgs(want.Name). WillReturnError(sql.ErrNoRows) }, @@ -151,7 +151,7 @@ func TestQueueStore_Get(t *testing.T) { name: "query error", queueName: want.Name, setup: func(mock sqlmock.Sqlmock) { - mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id FROM queue"). + mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms FROM queue"). WithArgs(want.Name). WillReturnError(fmt.Errorf("connection reset")) }, @@ -201,7 +201,7 @@ func TestQueueStore_Update(t *testing.T) { name: "success", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE queue"). - WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Name, oldVersion). + WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs, queue.Name, oldVersion). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -209,7 +209,7 @@ func TestQueueStore_Update(t *testing.T) { name: "version mismatch", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE queue"). - WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Name, oldVersion). + WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs, queue.Name, oldVersion). WillReturnResult(sqlmock.NewResult(0, 0)) }, wantErr: true, @@ -219,7 +219,7 @@ func TestQueueStore_Update(t *testing.T) { name: "exec error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE queue"). - WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Name, oldVersion). + WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs, queue.Name, oldVersion). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -228,7 +228,7 @@ func TestQueueStore_Update(t *testing.T) { name: "rows affected error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE queue"). - WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Name, oldVersion). + WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs, queue.Name, oldVersion). WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error"))) }, wantErr: true, diff --git a/stovepipe/extension/storage/mysql/queue_summary_store.go b/stovepipe/extension/storage/mysql/queue_summary_store.go new file mode 100644 index 000000000..fbb0a474a --- /dev/null +++ b/stovepipe/extension/storage/mysql/queue_summary_store.go @@ -0,0 +1,101 @@ +// 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/uber-go/tally" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +type queueSummaryStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +func NewQueueSummaryStore(db *sql.DB, scope tally.Scope, queue string) storage.QueueSummaryStore { + return &queueSummaryStore{db: db, scope: scope, queue: queue} +} + +func (s *queueSummaryStore) Get(ctx context.Context) (summary entity.QueueSummary, retErr error) { + op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + err := s.db.QueryRowContext(ctx, + "SELECT queue, state, effective_from_commit_uri, state_timestamp_ms, queue_state_version, version FROM queue_summary WHERE queue = ?", s.queue). + Scan(&summary.Queue, &summary.State, &summary.EffectiveFromCommitURI, &summary.StateTimestampMs, &summary.QueueStateVersion, &summary.Version) + if errors.Is(err, sql.ErrNoRows) { + return entity.QueueSummary{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.QueueSummary{}, fmt.Errorf("read queue summary %q: %w", s.queue, err) + } + return summary, nil +} + +func (s *queueSummaryStore) Create(ctx context.Context, summary entity.QueueSummary) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + if err := s.validateSummary(summary); err != nil { + return err + } + _, err := s.db.ExecContext(ctx, + "INSERT INTO queue_summary (queue, state, effective_from_commit_uri, state_timestamp_ms, queue_state_version, version) VALUES (?, ?, ?, ?, ?, ?)", + summary.Queue, summary.State, summary.EffectiveFromCommitURI, summary.StateTimestampMs, summary.QueueStateVersion, summary.Version) + if isDuplicateEntry(err) { + return storage.ErrAlreadyExists + } + if err != nil { + return fmt.Errorf("create queue summary %q: %w", s.queue, err) + } + return nil +} + +func (s *queueSummaryStore) Update(ctx context.Context, summary entity.QueueSummary, oldVersion, newVersion int64) (retErr error) { + op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + stored := summary + stored.Version = newVersion + if err := s.validateSummary(stored); err != nil { + return err + } + result, err := s.db.ExecContext(ctx, + "UPDATE queue_summary SET state = ?, effective_from_commit_uri = ?, state_timestamp_ms = ?, queue_state_version = ?, version = ? WHERE queue = ? AND version = ?", + summary.State, summary.EffectiveFromCommitURI, summary.StateTimestampMs, summary.QueueStateVersion, newVersion, s.queue, oldVersion) + if err != nil { + return fmt.Errorf("update queue summary %q: %w", s.queue, err) + } + count, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("read queue summary update count: %w", err) + } + if count != 1 { + return storage.ErrVersionMismatch + } + return nil +} + +func (s *queueSummaryStore) validateSummary(summary entity.QueueSummary) error { + if summary.Queue != s.queue { + return fmt.Errorf("summary queue differs from bound queue %q", s.queue) + } + return summary.Validate() +} diff --git a/stovepipe/extension/storage/mysql/queue_summary_store_test.go b/stovepipe/extension/storage/mysql/queue_summary_store_test.go new file mode 100644 index 000000000..269972fe2 --- /dev/null +++ b/stovepipe/extension/storage/mysql/queue_summary_store_test.go @@ -0,0 +1,190 @@ +// 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" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + drivermysql "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func queueSummaryFixture() entity.QueueSummary { + return entity.QueueSummary{Queue: "repo/main", State: entity.QueueEnablementEnabled, EffectiveFromCommitURI: "A", StateTimestampMs: 1000, Version: 2, QueueStateVersion: 7} +} + +func TestQueueSummaryStoreConditionalUpdate(t *testing.T) { + failure := errors.New("storage failure") + for _, tc := range []struct { + name string + setup func(sqlmock.Sqlmock) + want error + }{ + {"supplied versions", func(m sqlmock.Sqlmock) { + m.ExpectExec("UPDATE queue_summary").WithArgs("enabled", "A", int64(1000), int64(7), int64(9), "repo/main", int64(2)).WillReturnResult(sqlmock.NewResult(0, 1)) + }, nil}, + {"stale version", func(m sqlmock.Sqlmock) { + m.ExpectExec("UPDATE queue_summary").WillReturnResult(sqlmock.NewResult(0, 0)) + }, storage.ErrVersionMismatch}, + {"dependency failure", func(m sqlmock.Sqlmock) { m.ExpectExec("UPDATE queue_summary").WillReturnError(failure) }, failure}, + {"rows affected failure", func(m sqlmock.Sqlmock) { + m.ExpectExec("UPDATE queue_summary").WillReturnResult(sqlmock.NewErrorResult(failure)) + }, failure}, + } { + t.Run(tc.name, func(t *testing.T) { + db, m, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + tc.setup(m) + state := queueSummaryFixture() + err = NewQueueSummaryStore(db, tally.NoopScope, state.Queue).Update(context.Background(), state, 2, 9) + if tc.want == nil { + require.NoError(t, err) + } else { + require.ErrorIs(t, err, tc.want) + } + require.Equal(t, int64(2), state.Version) + require.NoError(t, m.ExpectationsWereMet()) + }) + } +} + +func TestQueueSummaryStoreCreateAndGet(t *testing.T) { + state := queueSummaryFixture() + failure := errors.New("storage failure") + for _, tc := range []struct { + name string + createErr error + getErr error + wantCreate error + wantGet error + }{ + {"round trip", nil, nil, nil, nil}, + {"duplicate", &drivermysql.MySQLError{Number: mysqlErrDuplicateEntry}, nil, storage.ErrAlreadyExists, nil}, + {"missing", nil, sql.ErrNoRows, nil, storage.ErrNotFound}, + {"dependency", failure, failure, failure, failure}, + } { + t.Run(tc.name, func(t *testing.T) { + db, m, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + exec := m.ExpectExec("INSERT INTO queue_summary").WithArgs(state.Queue, "enabled", "A", int64(1000), int64(7), int64(2)) + if tc.createErr != nil { + exec.WillReturnError(tc.createErr) + } else { + exec.WillReturnResult(sqlmock.NewResult(0, 1)) + } + query := m.ExpectQuery("SELECT .* FROM queue_summary WHERE queue").WithArgs(state.Queue) + if tc.getErr != nil { + query.WillReturnError(tc.getErr) + } else { + query.WillReturnRows(sqlmock.NewRows([]string{"queue", "state", "uri", "time", "state_version", "version"}).AddRow(state.Queue, state.State, "A", 1000, 7, 2)) + } + store := NewQueueSummaryStore(db, tally.NoopScope, state.Queue) + err = store.Create(context.Background(), state) + if tc.wantCreate != nil { + require.ErrorIs(t, err, tc.wantCreate) + } else { + require.NoError(t, err) + } + got, err := store.Get(context.Background()) + if tc.wantGet != nil { + require.ErrorIs(t, err, tc.wantGet) + } else { + require.NoError(t, err) + require.Equal(t, state, got) + } + require.NoError(t, m.ExpectationsWereMet()) + }) + } +} + +func TestQueueLogStoreImmutableOccurrence(t *testing.T) { + log := entity.QueueLog{ID: entity.QueueLogID(2), Queue: "repo/main", TimestampMs: 1000, State: entity.QueueEnablementEnabled, QueueStateVersion: 2, EffectiveFromCommitURI: "A"} + failure := errors.New("storage failure") + for _, tc := range []struct { + name string + createErr error + getErr error + wantCreate error + wantGet error + }{ + {"round trip", nil, nil, nil, nil}, + {"immutable duplicate", &drivermysql.MySQLError{Number: mysqlErrDuplicateEntry}, nil, storage.ErrAlreadyExists, nil}, + {"missing", nil, sql.ErrNoRows, nil, storage.ErrNotFound}, + {"dependency", failure, failure, failure, failure}, + } { + t.Run(tc.name, func(t *testing.T) { + db, m, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + exec := m.ExpectExec("INSERT INTO queue_log").WithArgs(log.Queue, log.ID, int64(1000), "enabled", int64(2), "A") + if tc.createErr != nil { + exec.WillReturnError(tc.createErr) + } else { + exec.WillReturnResult(sqlmock.NewResult(0, 1)) + } + query := m.ExpectQuery("SELECT .* FROM queue_log WHERE queue").WithArgs(log.Queue, log.ID) + if tc.getErr != nil { + query.WillReturnError(tc.getErr) + } else { + query.WillReturnRows(sqlmock.NewRows([]string{"queue", "id", "time", "state", "version", "uri"}).AddRow(log.Queue, log.ID, 1000, "enabled", 2, "A")) + } + store := NewQueueLogStore(db, tally.NoopScope, log.Queue) + err = store.Create(context.Background(), log) + if tc.wantCreate != nil { + require.ErrorIs(t, err, tc.wantCreate) + } else { + require.NoError(t, err) + } + got, err := store.Get(context.Background(), log.ID) + if tc.wantGet != nil { + require.ErrorIs(t, err, tc.wantGet) + } else { + require.NoError(t, err) + require.Equal(t, log, got) + } + require.NoError(t, m.ExpectationsWereMet()) + }) + } +} + +func TestQueueSummaryStoresRejectInvalidWrites(t *testing.T) { + db, m, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + state := queueSummaryFixture() + state.Queue = "other" + store := NewQueueSummaryStore(db, tally.NoopScope, "repo/main") + require.Error(t, store.Create(context.Background(), state)) + require.Error(t, store.Update(context.Background(), state, 1, 2)) + state.Queue = "repo/main" + state.State = entity.QueueEnablementUnknown + require.Error(t, store.Create(context.Background(), state)) + require.Error(t, store.Update(context.Background(), state, 1, 2)) + logs := NewQueueLogStore(db, tally.NoopScope, "repo/main") + log := entity.QueueLog{Queue: "other"} + require.Error(t, logs.Create(context.Background(), log)) + log.Queue = "repo/main" + require.Error(t, logs.Create(context.Background(), log)) + require.NoError(t, m.ExpectationsWereMet()) +} diff --git a/stovepipe/extension/storage/mysql/schema/queue.sql b/stovepipe/extension/storage/mysql/schema/queue.sql index c10463f0f..05acbc350 100644 --- a/stovepipe/extension/storage/mysql/schema/queue.sql +++ b/stovepipe/extension/storage/mysql/schema/queue.sql @@ -7,5 +7,9 @@ CREATE TABLE IF NOT EXISTS queue ( latest_request_id VARCHAR(255) NOT NULL DEFAULT '', version INT NOT NULL, last_green_request_id VARCHAR(255) NOT NULL DEFAULT '', + enablement VARCHAR(32) NOT NULL DEFAULT '', + enablement_version BIGINT NOT NULL DEFAULT 0, + effective_from_commit_uri VARCHAR(255) NOT NULL DEFAULT '', + enablement_changed_at_ms BIGINT NOT NULL DEFAULT 0, PRIMARY KEY (name) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/stovepipe/extension/storage/mysql/schema/queue_log.sql b/stovepipe/extension/storage/mysql/schema/queue_log.sql new file mode 100644 index 000000000..a00a0cedf --- /dev/null +++ b/stovepipe/extension/storage/mysql/schema/queue_log.sql @@ -0,0 +1,11 @@ +-- queue_log retains immutable successful state occurrences. State version determines +-- ordering and derives log identity, so every history read can use the primary key. +CREATE TABLE IF NOT EXISTS queue_log ( + queue VARBINARY(255) NOT NULL, + log_id VARBINARY(255) NOT NULL, + timestamp_ms BIGINT NOT NULL, + state VARCHAR(32) NOT NULL, + queue_state_version BIGINT NOT NULL, + effective_from_commit_uri VARBINARY(255) NOT NULL, + PRIMARY KEY (queue, log_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/stovepipe/extension/storage/mysql/schema/queue_summary.sql b/stovepipe/extension/storage/mysql/schema/queue_summary.sql new file mode 100644 index 000000000..b4e9e1307 --- /dev/null +++ b/stovepipe/extension/storage/mysql/schema/queue_summary.sql @@ -0,0 +1,10 @@ +-- queue_summary is the current read model projected from retained queue logs. +CREATE TABLE IF NOT EXISTS queue_summary ( + queue VARBINARY(255) NOT NULL, + state VARCHAR(32) NOT NULL, + effective_from_commit_uri VARCHAR(255) NOT NULL, + state_timestamp_ms BIGINT NOT NULL, + queue_state_version BIGINT NOT NULL, + version BIGINT NOT NULL, + PRIMARY KEY (queue) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; diff --git a/stovepipe/extension/storage/mysql/storage.go b/stovepipe/extension/storage/mysql/storage.go index a5369dfcc..e3961c95a 100644 --- a/stovepipe/extension/storage/mysql/storage.go +++ b/stovepipe/extension/storage/mysql/storage.go @@ -51,6 +51,8 @@ func (s *Storage) For(queueName string) (storage.Storage, error) { requestLogStore: NewRequestLogStore(s.db, queueScope.SubScope("request_log_store"), queueName), requestSummaryStore: NewRequestSummaryStore(s.db, queueScope.SubScope("request_summary_store"), queueName), requestAcceptanceStore: NewRequestAcceptanceStore(s.db, queueScope.SubScope("request_acceptance_store"), queueName), + queueSummaryStore: NewQueueSummaryStore(s.db, queueScope.SubScope("queue_summary_store"), queueName), + queueLogStore: NewQueueLogStore(s.db, queueScope.SubScope("queue_log_store"), queueName), queueStore: NewQueueStore(s.db, queueScope.SubScope("queue_store"), queueName), buildStore: NewBuildStore(s.db, queueScope.SubScope("build_store"), queueName), validationFactStore: NewValidationFactStore(s.db, queueScope.SubScope("validation_fact_store"), queueName), @@ -70,6 +72,8 @@ type mysqlStorage struct { requestSummaryStore storage.RequestSummaryStore requestAcceptanceStore storage.RequestAcceptanceStore queueStore storage.QueueStore + queueSummaryStore storage.QueueSummaryStore + queueLogStore storage.QueueLogStore buildStore storage.BuildStore validationFactStore storage.ValidationFactStore } @@ -116,3 +120,9 @@ func (f *mysqlStorage) GetBuildStore() storage.BuildStore { func (f *mysqlStorage) GetValidationFactStore() storage.ValidationFactStore { return f.validationFactStore } + +func (f *mysqlStorage) GetQueueSummaryStore() storage.QueueSummaryStore { return f.queueSummaryStore } + +func (f *mysqlStorage) GetQueueLogStore() storage.QueueLogStore { + return f.queueLogStore +} diff --git a/stovepipe/extension/storage/queue_log_store.go b/stovepipe/extension/storage/queue_log_store.go new file mode 100644 index 000000000..2336c1b0c --- /dev/null +++ b/stovepipe/extension/storage/queue_log_store.go @@ -0,0 +1,31 @@ +// 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=queue_log_store.go -destination=mock/queue_log_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// QueueLogStore retains immutable state occurrences for its bound queue. +type QueueLogStore interface { + // Get returns an occurrence by log ID or ErrNotFound. + Get(ctx context.Context, logID string) (entity.QueueLog, error) + // Create retains an occurrence, returning ErrAlreadyExists when its identity exists. + Create(context.Context, entity.QueueLog) error +} diff --git a/stovepipe/extension/storage/queue_summary_store.go b/stovepipe/extension/storage/queue_summary_store.go new file mode 100644 index 000000000..4b72ce053 --- /dev/null +++ b/stovepipe/extension/storage/queue_summary_store.go @@ -0,0 +1,33 @@ +// 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=queue_summary_store.go -destination=mock/queue_summary_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// QueueSummaryStore persists derived queue summaries for its bound queue. +type QueueSummaryStore interface { + // Get returns the current summary or ErrNotFound when no projection exists. + Get(context.Context) (entity.QueueSummary, error) + // Create creates a summary, returning ErrAlreadyExists for an existing queue. + Create(context.Context, entity.QueueSummary) error + // Update conditionally replaces the summary, guarding oldVersion and writing newVersion without arithmetic. + Update(ctx context.Context, summary entity.QueueSummary, oldVersion, newVersion int64) error +} diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index 692db6449..f0bb5d079 100644 --- a/stovepipe/extension/storage/storage.go +++ b/stovepipe/extension/storage/storage.go @@ -87,6 +87,10 @@ type Storage interface { // GetQueueStore returns the QueueStore instance. GetQueueStore() QueueStore + GetQueueSummaryStore() QueueSummaryStore + + GetQueueLogStore() QueueLogStore + // GetBuildStore returns the BuildStore instance. GetBuildStore() BuildStore diff --git a/test/e2e/stovepipe/suite_test.go b/test/e2e/stovepipe/suite_test.go index e0a9e74a9..79136c2a4 100644 --- a/test/e2e/stovepipe/suite_test.go +++ b/test/e2e/stovepipe/suite_test.go @@ -76,7 +76,7 @@ func (s *StovepipeE2ESuite) SetupSuite() { "service/stovepipe/server/Dockerfile": "service/stovepipe/server/Dockerfile", })) - err := s.stack.Up() + err := s.stack.Up("mysql-app", "mysql-queue") require.NoError(t, err, "failed to start compose stack") s.db, err = s.stack.ConnectMySQLService("mysql-app") @@ -85,14 +85,16 @@ func (s *StovepipeE2ESuite) SetupSuite() { s.queueDB, err = s.stack.ConnectMySQLService("mysql-queue") require.NoError(t, err, "failed to connect to queue MySQL") - // Apply schemas after the stack is up; the service connects lazily and the - // consumer retries, so the boot ordering is tolerated. + // Queue enablement is registered at service startup, so its schema must exist first. testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("platform/extension/counter/mysql/schema")) testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("stovepipe/extension/storage/mysql/schema")) testutil.ApplySchema(t, s.log, s.queueDB, testutil.SchemaDir("platform/extension/messagequeue/mysql/schema")) s.appStorage, err = storagemysql.NewStorage(s.db, tally.NoopScope) require.NoError(t, err) + err = s.stack.Up() + require.NoError(t, err, "failed to start service after applying schemas") + var conn *grpc.ClientConn conn, err = s.stack.ConnectGRPC("stovepipe-service", 8080) require.NoError(t, err, "failed to connect to stovepipe service") diff --git a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel index a12334038..408c73051 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -3,6 +3,8 @@ load("@rules_go//go:def.bzl", "go_test") go_test( name = "go_default_test", srcs = [ + "queue_state_transitions_test.go", + "queue_summary_test.go", "request_acceptance_test.go", "storage_test.go", ], @@ -15,8 +17,12 @@ go_test( "requires-network", ], deps = [ + "//stovepipe/core/queuelog:go_default_library", + "//stovepipe/core/queuestate:go_default_library", "//stovepipe/core/requestlog:go_default_library", "//stovepipe/entity:go_default_library", + "//stovepipe/extension/sourcecontrol:go_default_library", + "//stovepipe/extension/sourcecontrol/fake:go_default_library", "//stovepipe/extension/storage:go_default_library", "//stovepipe/extension/storage/mysql:go_default_library", "//test/integration/stovepipe/extension/storage:go_default_library", diff --git a/test/integration/stovepipe/extension/storage/mysql/queue_state_transitions_test.go b/test/integration/stovepipe/extension/storage/mysql/queue_state_transitions_test.go new file mode 100644 index 000000000..fc127fccc --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/queue_state_transitions_test.go @@ -0,0 +1,71 @@ +// 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" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/stovepipe/core/queuelog" + "github.com/uber/submitqueue/stovepipe/core/queuestate" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" + scfake "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/fake" + mysqlstorage "github.com/uber/submitqueue/stovepipe/extension/storage/mysql" +) + +type transitionSourceFactory struct{ sc sourcecontrol.SourceControl } + +func (f *transitionSourceFactory) For(sourcecontrol.Config) (sourcecontrol.SourceControl, error) { + return f.sc, nil +} + +func testQueueStateTransitions(t *testing.T, ctx context.Context, db *sql.DB, backend *mysqlstorage.Storage) { + t.Helper() + const queue = "transition/main" + source := &transitionSourceFactory{sc: scfake.New(sourcecontrol.Config{QueueName: queue}, []string{"A", "before"})} + manager := queuestate.NewTransitionController(mysqlFactory{backend: backend}, source, queuelog.NewMaterializer(tally.NoopScope)) + _, err := manager.Initialize(ctx, queue) + require.NoError(t, err) + enable, err := manager.Prepare(ctx, queue, entity.QueueEnablementEnabled) + require.NoError(t, err) + enabled, err := manager.Apply(ctx, enable) + require.NoError(t, err) + require.Equal(t, int64(2), enabled.Version) + // Simulate a crash after the snapshot commit but before history retention. + _, err = db.ExecContext(ctx, "DELETE FROM queue_log WHERE queue = ? AND log_id = ?", queue, entity.QueueLogID(2)) + require.NoError(t, err) + source.sc = scfake.New(sourcecontrol.Config{QueueName: queue}, []string{"B", "A", "before"}) + disable, err := manager.Prepare(ctx, queue, entity.QueueEnablementDisabled) + require.NoError(t, err) + bound, err := backend.For(queue) + require.NoError(t, err) + repaired, err := bound.GetQueueLogStore().Get(ctx, entity.QueueLogID(2)) + require.NoError(t, err) + require.Equal(t, enabled, repaired.Snapshot()) + disabled, err := manager.Apply(ctx, disable) + require.NoError(t, err) + require.Equal(t, int64(3), disabled.Version) + restarted := queuestate.NewTransitionController(mysqlFactory{backend: backend}, source, queuelog.NewMaterializer(tally.NoopScope)) + replay, err := restarted.Apply(ctx, enable) + require.NoError(t, err) + require.Equal(t, enabled, replay) + current, err := bound.GetQueueSummaryStore().Get(ctx) + require.NoError(t, err) + require.Equal(t, disabled, current.Snapshot()) +} diff --git a/test/integration/stovepipe/extension/storage/mysql/queue_summary_test.go b/test/integration/stovepipe/extension/storage/mysql/queue_summary_test.go new file mode 100644 index 000000000..6dfc6334a --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/queue_summary_test.go @@ -0,0 +1,122 @@ +// 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" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + mysqlstorage "github.com/uber/submitqueue/stovepipe/extension/storage/mysql" +) + +func testQueueSummaryStores(t *testing.T, ctx context.Context, backend *mysqlstorage.Storage) { + t.Helper() + bound, err := backend.For("state/Main") + require.NoError(t, err) + other, err := backend.For("state/main") + require.NoError(t, err) + states := bound.GetQueueSummaryStore() + logs := bound.GetQueueLogStore() + _, err = states.Get(ctx) + require.ErrorIs(t, err, storage.ErrNotFound) + state := entity.QueueSummary{Queue: "state/Main", State: entity.QueueEnablementDisabled, Version: 1, QueueStateVersion: 1, StateTimestampMs: 1000} + require.NoError(t, states.Create(ctx, state)) + require.ErrorIs(t, states.Create(ctx, state), storage.ErrAlreadyExists) + _, err = other.GetQueueSummaryStore().Get(ctx) + require.ErrorIs(t, err, storage.ErrNotFound) + state.State = entity.QueueEnablementEnabled + state.EffectiveFromCommitURI = "A" + state.StateTimestampMs = 2000 + state.QueueStateVersion = 2 + require.NoError(t, states.Update(ctx, state, 1, 2)) + state.Version = 2 + got, err := states.Get(ctx) + require.NoError(t, err) + require.Equal(t, state, got) + losing := state + losing.State = entity.QueueEnablementDisabled + losing.EffectiveFromCommitURI = "B" + require.ErrorIs(t, states.Update(ctx, losing, 1, 3), storage.ErrVersionMismatch) + got, err = states.Get(ctx) + require.NoError(t, err) + require.Equal(t, state, got) + log := entity.QueueLog{ID: entity.QueueLogID(2), Queue: state.Queue, State: state.State, TimestampMs: state.StateTimestampMs, QueueStateVersion: 2, EffectiveFromCommitURI: "A"} + require.NoError(t, logs.Create(ctx, log)) + conflicting := log + conflicting.State = entity.QueueEnablementDisabled + require.ErrorIs(t, logs.Create(ctx, conflicting), storage.ErrAlreadyExists) + retained, err := logs.Get(ctx, log.ID) + require.NoError(t, err) + require.Equal(t, log, retained) + _, err = other.GetQueueLogStore().Get(ctx, log.ID) + require.ErrorIs(t, err, storage.ErrNotFound) + require.Error(t, other.GetQueueLogStore().Create(ctx, log)) + _, err = logs.Get(ctx, entity.QueueLogID(1)) + require.ErrorIs(t, err, storage.ErrNotFound) +} + +func testQueueSummaryKeyIsolation(t *testing.T, ctx context.Context, backend *mysqlstorage.Storage) { + t.Helper() + for _, tt := range []struct { + name string + queue string + otherQueue string + }{ + {"case difference", "summary/case/main", "summary/case/Main"}, + {"trailing space", "summary/space/main", "summary/space/main "}, + } { + t.Run(tt.name, func(t *testing.T) { + bound, err := backend.For(tt.queue) + require.NoError(t, err) + other, err := backend.For(tt.otherQueue) + require.NoError(t, err) + store, otherStore := bound.GetQueueSummaryStore(), other.GetQueueSummaryStore() + initial := entity.QueueSummary{Queue: tt.queue, State: entity.QueueEnablementDisabled, Version: 1, QueueStateVersion: 1, StateTimestampMs: 1000} + require.NoError(t, store.Create(ctx, initial)) + _, err = otherStore.Get(ctx) + require.ErrorIs(t, err, storage.ErrNotFound) + otherState := initial + otherState.Queue = tt.otherQueue + require.NoError(t, otherStore.Create(ctx, otherState)) + otherState.State = entity.QueueEnablementEnabled + otherState.EffectiveFromCommitURI = "B" + otherState.StateTimestampMs = 2000 + otherState.QueueStateVersion = 2 + require.NoError(t, otherStore.Update(ctx, otherState, 1, 2)) + otherState.Version = 2 + got, err := store.Get(ctx) + require.NoError(t, err) + require.Equal(t, initial, got) + updated := initial + updated.State = entity.QueueEnablementEnabled + updated.EffectiveFromCommitURI = "A" + updated.StateTimestampMs = 3000 + updated.QueueStateVersion = 2 + require.NoError(t, store.Update(ctx, updated, 1, 2)) + updated.Version = 2 + require.ErrorIs(t, store.Update(ctx, initial, 1, 3), storage.ErrVersionMismatch) + require.ErrorIs(t, otherStore.Update(ctx, otherState, 1, 3), storage.ErrVersionMismatch) + got, err = store.Get(ctx) + require.NoError(t, err) + require.Equal(t, updated, got) + got, err = otherStore.Get(ctx) + require.NoError(t, err) + require.Equal(t, otherState, got) + }) + } +} diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index 7ed345b2f..fc93afcaf 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -54,6 +54,21 @@ func TestMySQLStorage(t *testing.T) { require.NoError(t, err, "failed to create storage") factory := mysqlFactory{backend: backend} + t.Run("QueueStateTransitions", func(t *testing.T) { + resetStorage(t, db) + testQueueStateTransitions(t, ctx, db, backend) + }) + + t.Run("QueueSummaryStores", func(t *testing.T) { + resetStorage(t, db) + testQueueSummaryStores(t, ctx, backend) + }) + + t.Run("QueueSummaryKeyIsolation", func(t *testing.T) { + resetStorage(t, db) + testQueueSummaryKeyIsolation(t, ctx, backend) + }) + t.Run("RequestStore", func(t *testing.T) { resetStorage(t, db) bound, err := backend.For("monorepo/main") @@ -128,6 +143,8 @@ func resetStorage(t *testing.T, db *sql.DB) { t.Helper() for _, statement := range []string{ + "TRUNCATE TABLE queue_log", + "TRUNCATE TABLE queue_summary", "TRUNCATE TABLE request_log", "TRUNCATE TABLE request_summary", "TRUNCATE TABLE request_acceptance", diff --git a/test/integration/stovepipe/suite_test.go b/test/integration/stovepipe/suite_test.go index f98d85159..7afc77bd0 100644 --- a/test/integration/stovepipe/suite_test.go +++ b/test/integration/stovepipe/suite_test.go @@ -73,7 +73,7 @@ func (s *StovepipeIntegrationSuite) SetupSuite() { "service/stovepipe/server/Dockerfile": "service/stovepipe/server/Dockerfile", })) - err := s.stack.Up() + err := s.stack.Up("mysql-app", "mysql-queue") require.NoError(t, err, "failed to start compose stack") s.db, err = s.stack.ConnectMySQLService("mysql-app") @@ -82,12 +82,14 @@ func (s *StovepipeIntegrationSuite) SetupSuite() { s.queueDB, err = s.stack.ConnectMySQLService("mysql-queue") require.NoError(t, err, "failed to connect to queue MySQL") - // Apply schemas after the stack is up; the service connects lazily and the - // consumer retries, so the boot ordering is tolerated. + // Queue enablement is registered at service startup, so its schema must exist first. testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("platform/extension/counter/mysql/schema")) testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("stovepipe/extension/storage/mysql/schema")) testutil.ApplySchema(t, s.log, s.queueDB, testutil.SchemaDir("platform/extension/messagequeue/mysql/schema")) + err = s.stack.Up() + require.NoError(t, err, "failed to start service after applying schemas") + var conn *grpc.ClientConn conn, err = s.stack.ConnectGRPC("stovepipe-service", 8080) require.NoError(t, err, "failed to connect to stovepipe service") diff --git a/test/testutil/compose.go b/test/testutil/compose.go index ebc2caf6a..a9efefa98 100644 --- a/test/testutil/compose.go +++ b/test/testutil/compose.go @@ -177,13 +177,13 @@ func NewComposeStack(t *testing.T, log *TestLogger, ctx context.Context, compose return stack } -// Up starts all services in the compose stack. +// Up starts the named services, or all services when none are supplied. // Uses --wait to block until all services with healthcheck directives are healthy. -func (s *ComposeStack) Up() error { +func (s *ComposeStack) Up(services ...string) error { s.t.Helper() s.log.Logf("Starting compose stack from %s", strings.Join(s.composeFiles, " + ")) - args := s.command("up", "-d", "--build", "--wait") + args := s.command(append([]string{"up", "-d", "--build", "--wait"}, services...)...) cmd := exec.CommandContext(s.ctx, s.composeCmd[0], args...) cmd.Env = s.composeEnv cmd.Stdout = os.Stdout @@ -194,6 +194,7 @@ func (s *ComposeStack) Up() error { } s.log.Logf("Compose stack started successfully") + s.stopLogs() s.tailLogs() return nil } diff --git a/web/test/e2e/stovepipe/run_e2e.py b/web/test/e2e/stovepipe/run_e2e.py index 9399ee2ab..7c7ec16ea 100644 --- a/web/test/e2e/stovepipe/run_e2e.py +++ b/web/test/e2e/stovepipe/run_e2e.py @@ -85,7 +85,20 @@ def main(): }, env=env, ) - run([*compose, "up", "-d", "--no-build", "--pull", "never", "--wait"], env) + run( + [ + *compose, + "up", + "-d", + "--no-build", + "--pull", + "never", + "--wait", + "mysql-app", + "mysql-queue", + ], + env, + ) apply_schemas( [ ( @@ -117,6 +130,7 @@ def main(): env, input=_pagination_fixture_sql(), ) + run([*compose, "up", "-d", "--no-build", "--pull", "never", "--wait"], env) for key, service, port in [ ("STOVEPIPE_E2E_WEB_URL", "web-service", "3000"), ("STOVEPIPE_E2E_URL", "stovepipe-service", "8080"), diff --git a/web/test/e2e/stovepipe/run_e2e_test.py b/web/test/e2e/stovepipe/run_e2e_test.py index 3978a944d..d8ef36bf8 100644 --- a/web/test/e2e/stovepipe/run_e2e_test.py +++ b/web/test/e2e/stovepipe/run_e2e_test.py @@ -9,9 +9,9 @@ from web.test.e2e.stovepipe.run_e2e import main -class RunnerCleanupTest(unittest.TestCase): - def test_termination_during_playwright_cleans_up_the_stack(self): - for signum in (signal.SIGTERM, signal.SIGINT): +class RunnerLifecycleTest(unittest.TestCase): + def test_startup_order_and_cleanup(self): + for signum in (None, signal.SIGTERM, signal.SIGINT): with self.subTest( signal=signum ), tempfile.TemporaryDirectory() as directory: @@ -25,10 +25,14 @@ def test_termination_during_playwright_cleans_up_the_stack(self): ): (test_dir / name).touch() commands = [] + schema_application = [] + + def record_schema_application(_groups, *, env): + schema_application.append(len(commands)) def run(command, **kwargs): commands.append(command) - if command[0] == str(root / "playwright"): + if command[0] == str(root / "playwright") and signum is not None: signal.raise_signal(signum) return subprocess.CompletedProcess(command, 0) @@ -58,14 +62,33 @@ def run(command, **kwargs): "subprocess.run", side_effect=run ), patch( "subprocess.check_output", return_value="127.0.0.1:12345" + ), patch( + "web.test.e2e.stovepipe.run_e2e.apply_schemas", + side_effect=record_schema_application, ): - with self.assertRaises(SystemExit) as exit: - main() - self.assertEqual(exit.exception.code, 128 + signum) + if signum is None: + self.assertEqual(main(), 0) + else: + with self.assertRaises(SystemExit) as exit: + main() + self.assertEqual(exit.exception.code, 128 + signum) finally: for sig, handler in handlers.items(): signal.signal(sig, handler) + startups = [ + (index, command) + for index, command in enumerate(commands) + if "up" in command + ] + self.assertEqual(len(startups), 2) + self.assertEqual(startups[0][1][-2:], ["mysql-app", "mysql-queue"]) + self.assertEqual(len(schema_application), 1) + self.assertLess(startups[0][0], schema_application[0]) + self.assertLess(schema_application[0], startups[1][0]) + self.assertEqual( + commands[schema_application[0]][:3], ["docker", "exec", "-i"] + ) self.assertTrue( any( command[-3:] == ["down", "-v", "--remove-orphans"]