From 5fcaa49ef4f5713969e4a81c273acf76eab00e90 Mon Sep 17 00:00:00 2001 From: manjari Date: Thu, 15 Oct 2026 22:01:29 +0000 Subject: [PATCH] refactor(conflict): Move target dictionary store to tango mysql --- .../conflict/tango/mysql/BUILD.bazel | 2 + .../conflict/tango/mysql/dictionary_store.go | 85 +++++++++ .../tango/mysql/dictionary_store_test.go | 164 ++++++++++++++++++ .../tango/mysql/schema/target_dictionary.sql | 9 + .../conflict/tango/mysql/BUILD.bazel | 5 +- .../tango/mysql/dictionary_store_test.go | 95 ++++++++++ .../tango/mysql/signature_store_test.go | 18 +- 7 files changed, 370 insertions(+), 8 deletions(-) create mode 100644 submitqueue/extension/conflict/tango/mysql/dictionary_store.go create mode 100644 submitqueue/extension/conflict/tango/mysql/dictionary_store_test.go create mode 100644 submitqueue/extension/conflict/tango/mysql/schema/target_dictionary.sql create mode 100644 test/integration/submitqueue/extension/conflict/tango/mysql/dictionary_store_test.go diff --git a/submitqueue/extension/conflict/tango/mysql/BUILD.bazel b/submitqueue/extension/conflict/tango/mysql/BUILD.bazel index cda15f7c7..34ca0204d 100644 --- a/submitqueue/extension/conflict/tango/mysql/BUILD.bazel +++ b/submitqueue/extension/conflict/tango/mysql/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", srcs = [ + "dictionary_store.go", "mysql.go", "signature_codec.go", "signature_store.go", @@ -25,6 +26,7 @@ go_library( go_test( name = "go_default_test", srcs = [ + "dictionary_store_test.go", "signature_codec_test.go", "signature_store_test.go", ], diff --git a/submitqueue/extension/conflict/tango/mysql/dictionary_store.go b/submitqueue/extension/conflict/tango/mysql/dictionary_store.go new file mode 100644 index 000000000..8aea41bc8 --- /dev/null +++ b/submitqueue/extension/conflict/tango/mysql/dictionary_store.go @@ -0,0 +1,85 @@ +// 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/go-sql-driver/mysql" + "github.com/uber-go/tally" + + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +type dictionaryStore struct { + db *sql.DB + scope tally.Scope + // queue is the queue name this store instance is bound to; every read and + // write is scoped to it. + queue string +} + +// NewDictionaryStore creates a new MySQL-backed TargetDictionaryStore. +func NewDictionaryStore(db *sql.DB, scope tally.Scope, queue string) tango.DictionaryStore { + return &dictionaryStore{db: db, scope: scope, queue: queue} +} + +// Create inserts a label. Returns ErrAlreadyExists if the ID is already recorded. +func (s *dictionaryStore) Create(ctx context.Context, label tango.TargetLabel) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if label.Queue != s.queue { + return fmt.Errorf("target label id=%d queue %q does not match the store's bound queue %q", label.ID, label.Queue, s.queue) + } + + _, err := s.db.ExecContext(ctx, + "INSERT INTO target_dictionary (queue, id, label, created_at) VALUES (?, ?, ?, ?)", + label.Queue, label.ID, label.Label, label.CreatedAt, + ) + if err != nil { + var mysqlErr *mysql.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry { + return fmt.Errorf("target label id=%d: %w", label.ID, storage.ErrAlreadyExists) + } + return fmt.Errorf("failed to insert target label id=%d: %w", label.ID, err) + } + return nil +} + +// Get returns the bound queue's label for id. The bool is false when none is +// recorded. +func (s *dictionaryStore) Get(ctx context.Context, id uint64) (ret tango.TargetLabel, found bool, retErr error) { + op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + label := tango.TargetLabel{Queue: s.queue, ID: id} + err := s.db.QueryRowContext(ctx, + "SELECT label, created_at FROM target_dictionary WHERE queue = ? AND id = ?", + s.queue, id, + ).Scan(&label.Label, &label.CreatedAt) + if errors.Is(err, sql.ErrNoRows) { + return tango.TargetLabel{}, false, nil + } + if err != nil { + return tango.TargetLabel{}, false, fmt.Errorf("failed to get target label id=%d from the database: %w", id, err) + } + return label, true, nil +} diff --git a/submitqueue/extension/conflict/tango/mysql/dictionary_store_test.go b/submitqueue/extension/conflict/tango/mysql/dictionary_store_test.go new file mode 100644 index 000000000..9ea3efd87 --- /dev/null +++ b/submitqueue/extension/conflict/tango/mysql/dictionary_store_test.go @@ -0,0 +1,164 @@ +// 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" + "fmt" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +func setupDictionaryStoreTest(t *testing.T) (*sql.DB, sqlmock.Sqlmock, tango.DictionaryStore) { + t.Helper() + + db, mock, err := sqlmock.New() + require.NoError(t, err) + + return db, mock, NewDictionaryStore(db, testMetrics(), "monorepo") +} + +func TestTargetDictionaryStore_Create(t *testing.T) { + label := tango.TargetLabel{Queue: "monorepo", ID: 42, Label: "//app/server:lib", CreatedAt: 1700000000000} + mismatched := label + mismatched.Queue = "other-queue" + + tests := []struct { + name string + label tango.TargetLabel + setup func(mock sqlmock.Sqlmock) + wantErr bool + wantErrIs error + }{ + { + name: "success", + label: label, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO target_dictionary"). + WithArgs("monorepo", uint64(42), label.Label, label.CreatedAt). + WillReturnResult(sqlmock.NewResult(1, 1)) + }, + }, + { + name: "id already recorded", + label: label, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO target_dictionary"). + WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) + }, + wantErr: true, + wantErrIs: storage.ErrAlreadyExists, + }, + { + name: "insert error", + label: label, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO target_dictionary"). + WillReturnError(fmt.Errorf("data too long for column label")) + }, + wantErr: true, + }, + { + name: "queue differs from the bound queue", + label: mismatched, + setup: func(mock sqlmock.Sqlmock) {}, + wantErr: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, store := setupDictionaryStoreTest(t) + defer db.Close() + tt.setup(mock) + + err := store.Create(context.Background(), tt.label) + + if tt.wantErr { + require.Error(t, err) + if tt.wantErrIs != nil { + assert.ErrorIs(t, err, tt.wantErrIs) + } + } else { + require.NoError(t, err) + } + assert.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestTargetDictionaryStore_Get(t *testing.T) { + columns := []string{"label", "created_at"} + + tests := []struct { + name string + setup func(mock sqlmock.Sqlmock) + want tango.TargetLabel + wantFound bool + wantErr bool + }{ + { + name: "found", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT label, created_at FROM target_dictionary"). + WithArgs("monorepo", uint64(42)). + WillReturnRows(sqlmock.NewRows(columns).AddRow("//app:lib", int64(10))) + }, + want: tango.TargetLabel{Queue: "monorepo", ID: 42, Label: "//app:lib", CreatedAt: 10}, + wantFound: true, + }, + { + name: "none recorded", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT label, created_at FROM target_dictionary"). + WithArgs("monorepo", uint64(42)). + WillReturnRows(sqlmock.NewRows(columns)) + }, + }, + { + name: "query error", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT label, created_at FROM target_dictionary"). + WillReturnError(fmt.Errorf("connection reset")) + }, + wantErr: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, store := setupDictionaryStoreTest(t) + defer db.Close() + tt.setup(mock) + + got, found, err := store.Get(context.Background(), 42) + + if tt.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + assert.Equal(t, tt.wantFound, found) + assert.Equal(t, tt.want, got) + } + assert.NoError(t, mock.ExpectationsWereMet()) + }) + } +} diff --git a/submitqueue/extension/conflict/tango/mysql/schema/target_dictionary.sql b/submitqueue/extension/conflict/tango/mysql/schema/target_dictionary.sql new file mode 100644 index 000000000..0aecaae6f --- /dev/null +++ b/submitqueue/extension/conflict/tango/mysql/schema/target_dictionary.sql @@ -0,0 +1,9 @@ +-- Rows are written once and never updated. Two labels can share an id on a +-- hash collision; the first one written stays. +CREATE TABLE IF NOT EXISTS target_dictionary ( + queue VARCHAR(255) NOT NULL, + id BIGINT UNSIGNED NOT NULL, + label TEXT NOT NULL, + created_at BIGINT NOT NULL, + PRIMARY KEY (queue, id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/test/integration/submitqueue/extension/conflict/tango/mysql/BUILD.bazel b/test/integration/submitqueue/extension/conflict/tango/mysql/BUILD.bazel index c6ed1fbf0..77345e8f3 100644 --- a/test/integration/submitqueue/extension/conflict/tango/mysql/BUILD.bazel +++ b/test/integration/submitqueue/extension/conflict/tango/mysql/BUILD.bazel @@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_test") go_test( name = "go_default_test", - srcs = ["signature_store_test.go"], + srcs = [ + "dictionary_store_test.go", + "signature_store_test.go", + ], data = [ "docker-compose.yml", "//submitqueue/extension/conflict/tango/mysql/schema", diff --git a/test/integration/submitqueue/extension/conflict/tango/mysql/dictionary_store_test.go b/test/integration/submitqueue/extension/conflict/tango/mysql/dictionary_store_test.go new file mode 100644 index 000000000..3f8457a2e --- /dev/null +++ b/test/integration/submitqueue/extension/conflict/tango/mysql/dictionary_store_test.go @@ -0,0 +1,95 @@ +// 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" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +func (s *StoreIntegrationSuite) TestDictionaryStore() { + t := s.T() + ctx := context.Background() + storeA := s.forQueueDictionary("dict-queue-a") + storeB := s.forQueueDictionary("dict-queue-b") + + // An ID with the high bit set must survive the unsigned column round trip. + const highID = uint64(1) << 63 + label := func(id uint64, name string) tango.TargetLabel { + return tango.TargetLabel{Queue: "dict-queue-a", ID: id, Label: name, CreatedAt: 1700000000000} + } + + t.Run("round trip", func(t *testing.T) { + require.NoError(t, storeA.Create(ctx, label(7, "//app/server:lib"))) + + got, found, err := storeA.Get(ctx, 7) + require.NoError(t, err) + assert.True(t, found) + assert.Equal(t, label(7, "//app/server:lib"), got) + }) + + t.Run("ID with the high bit set", func(t *testing.T) { + require.NoError(t, storeA.Create(ctx, label(highID, "//high:bit"))) + + got, found, err := storeA.Get(ctx, highID) + require.NoError(t, err) + assert.True(t, found) + assert.Equal(t, label(highID, "//high:bit"), got) + }) + + t.Run("long label", func(t *testing.T) { + long := "//" + strings.Repeat("p", 900) + ":target" + require.NoError(t, storeA.Create(ctx, label(11, long))) + + got, found, err := storeA.Get(ctx, 11) + require.NoError(t, err) + assert.True(t, found) + assert.Equal(t, label(11, long), got) + }) + + t.Run("a second label for a recorded ID is rejected and the first stays", func(t *testing.T) { + require.NoError(t, storeA.Create(ctx, label(9, "//a:x"))) + require.ErrorIs(t, storeA.Create(ctx, label(9, "//b:y")), storage.ErrAlreadyExists) + + got, found, err := storeA.Get(ctx, 9) + require.NoError(t, err) + assert.True(t, found) + assert.Equal(t, label(9, "//a:x"), got) + }) + + t.Run("repeated create returns ErrAlreadyExists", func(t *testing.T) { + require.ErrorIs(t, storeA.Create(ctx, label(7, "//app/server:lib")), storage.ErrAlreadyExists) + }) + + t.Run("unknown ID is not found", func(t *testing.T) { + _, found, err := storeA.Get(ctx, 12345) + require.NoError(t, err) + assert.False(t, found) + }) + + t.Run("invisible through another queue", func(t *testing.T) { + _, found, err := storeB.Get(ctx, 7) + require.NoError(t, err) + assert.False(t, found) + require.Error(t, storeB.Create(ctx, label(7, "//app/server:lib")), "a mismatched-queue write must be rejected") + }) +} diff --git a/test/integration/submitqueue/extension/conflict/tango/mysql/signature_store_test.go b/test/integration/submitqueue/extension/conflict/tango/mysql/signature_store_test.go index 9f47e8637..f3caf62d2 100644 --- a/test/integration/submitqueue/extension/conflict/tango/mysql/signature_store_test.go +++ b/test/integration/submitqueue/extension/conflict/tango/mysql/signature_store_test.go @@ -31,20 +31,20 @@ import ( "github.com/uber/submitqueue/test/testutil" ) -// SignatureStoreIntegrationSuite runs the MySQL signature store against a real +// StoreIntegrationSuite runs the MySQL stores against a real // database. -type SignatureStoreIntegrationSuite struct { +type StoreIntegrationSuite struct { suite.Suite stack *testutil.ComposeStack db *sql.DB log *testutil.TestLogger } -func TestSignatureStoreIntegration(t *testing.T) { - suite.Run(t, new(SignatureStoreIntegrationSuite)) +func TestStoreIntegration(t *testing.T) { + suite.Run(t, new(StoreIntegrationSuite)) } -func (s *SignatureStoreIntegrationSuite) SetupSuite() { +func (s *StoreIntegrationSuite) SetupSuite() { t := s.T() ctx := context.Background() s.log = testutil.NewTestLogger(t) @@ -60,11 +60,15 @@ func (s *SignatureStoreIntegrationSuite) SetupSuite() { t.Cleanup(func() { s.db.Close() }) } -func (s *SignatureStoreIntegrationSuite) forQueue(queue string) tango.SignatureStore { +func (s *StoreIntegrationSuite) forQueue(queue string) tango.SignatureStore { return tangomysql.NewSignatureStore(s.db, tally.NoopScope, queue) } -func (s *SignatureStoreIntegrationSuite) TestSignatureStore() { +func (s *StoreIntegrationSuite) forQueueDictionary(queue string) tango.DictionaryStore { + return tangomysql.NewDictionaryStore(s.db, tally.NoopScope, queue) +} + +func (s *StoreIntegrationSuite) TestSignatureStore() { t := s.T() ctx := context.Background() storeA := s.forQueue("sig-queue-a")