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/integration/stovepipe/extension/storage/mysql/BUILD.bazel b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel index a12334038..83f10df7f 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_test") go_test( name = "go_default_test", srcs = [ + "queue_summary_test.go", "request_acceptance_test.go", "storage_test.go", ], 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..71004674c 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -54,6 +54,16 @@ func TestMySQLStorage(t *testing.T) { require.NoError(t, err, "failed to create storage") factory := mysqlFactory{backend: 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 +138,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",