From d949a1cab00081b1c8decba36d7b2510debdee37 Mon Sep 17 00:00:00 2001 From: manjari Date: Thu, 15 Oct 2026 21:50:25 +0000 Subject: [PATCH 1/2] refactor(conflict): Move target signature store to tango mysql --- .../conflict/tango/mysql/BUILD.bazel | 41 ++++ .../extension/conflict/tango/mysql/mysql.go | 19 ++ .../conflict/tango/mysql/schema/BUILD.bazel | 5 + .../tango/mysql/schema/target_signature.sql | 11 + .../conflict/tango/mysql/signature_codec.go | 64 +++++ .../tango/mysql/signature_codec_test.go | 116 +++++++++ .../conflict/tango/mysql/signature_store.go | 91 +++++++ .../tango/mysql/signature_store_test.go | 223 ++++++++++++++++++ .../conflict/tango/mysql/BUILD.bazel | 25 ++ .../conflict/tango/mysql/docker-compose.yml | 24 ++ .../tango/mysql/signature_store_test.go | 132 +++++++++++ tool/linter/queueshard/main.go | 2 + 12 files changed, 753 insertions(+) create mode 100644 submitqueue/extension/conflict/tango/mysql/BUILD.bazel create mode 100644 submitqueue/extension/conflict/tango/mysql/mysql.go create mode 100644 submitqueue/extension/conflict/tango/mysql/schema/BUILD.bazel create mode 100644 submitqueue/extension/conflict/tango/mysql/schema/target_signature.sql create mode 100644 submitqueue/extension/conflict/tango/mysql/signature_codec.go create mode 100644 submitqueue/extension/conflict/tango/mysql/signature_codec_test.go create mode 100644 submitqueue/extension/conflict/tango/mysql/signature_store.go create mode 100644 submitqueue/extension/conflict/tango/mysql/signature_store_test.go create mode 100644 test/integration/submitqueue/extension/conflict/tango/mysql/BUILD.bazel create mode 100644 test/integration/submitqueue/extension/conflict/tango/mysql/docker-compose.yml create mode 100644 test/integration/submitqueue/extension/conflict/tango/mysql/signature_store_test.go diff --git a/submitqueue/extension/conflict/tango/mysql/BUILD.bazel b/submitqueue/extension/conflict/tango/mysql/BUILD.bazel new file mode 100644 index 000000000..cda15f7c7 --- /dev/null +++ b/submitqueue/extension/conflict/tango/mysql/BUILD.bazel @@ -0,0 +1,41 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = [ + "mysql.go", + "signature_codec.go", + "signature_store.go", + ], + importpath = "github.com/uber/submitqueue/submitqueue/extension/conflict/tango/mysql", + visibility = [ + "//service/submitqueue/orchestrator:__subpackages__", + "//submitqueue/extension/conflict/tango:__subpackages__", + "//test/integration/submitqueue/extension/conflict/tango:__subpackages__", + ], + deps = [ + "//platform/metrics:go_default_library", + "//submitqueue/extension/conflict/tango:go_default_library", + "//submitqueue/extension/storage:go_default_library", + "@com_github_go_sql_driver_mysql//:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = [ + "signature_codec_test.go", + "signature_store_test.go", + ], + embed = [":go_default_library"], + deps = [ + "//submitqueue/extension/conflict/tango:go_default_library", + "//submitqueue/extension/storage:go_default_library", + "@com_github_data_dog_go_sqlmock//:go_default_library", + "@com_github_go_sql_driver_mysql//:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + ], +) diff --git a/submitqueue/extension/conflict/tango/mysql/mysql.go b/submitqueue/extension/conflict/tango/mysql/mysql.go new file mode 100644 index 000000000..4a0d0ddad --- /dev/null +++ b/submitqueue/extension/conflict/tango/mysql/mysql.go @@ -0,0 +1,19 @@ +// 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 implements the tango analyzer's stores over MySQL. +package mysql + +// mysqlErrDuplicateEntry is MySQL error code 1062 ("Duplicate entry"), returned on a unique or primary key violation. +const mysqlErrDuplicateEntry = 1062 diff --git a/submitqueue/extension/conflict/tango/mysql/schema/BUILD.bazel b/submitqueue/extension/conflict/tango/mysql/schema/BUILD.bazel new file mode 100644 index 000000000..3412d7733 --- /dev/null +++ b/submitqueue/extension/conflict/tango/mysql/schema/BUILD.bazel @@ -0,0 +1,5 @@ +filegroup( + name = "schema", + srcs = glob(["*.sql"]), + visibility = ["//visibility:public"], +) diff --git a/submitqueue/extension/conflict/tango/mysql/schema/target_signature.sql b/submitqueue/extension/conflict/tango/mysql/schema/target_signature.sql new file mode 100644 index 000000000..e7e83d652 --- /dev/null +++ b/submitqueue/extension/conflict/tango/mysql/schema/target_signature.sql @@ -0,0 +1,11 @@ +-- targets is the binary encoding of the signature's sorted (id, distance) +-- entries. LONGBLOB because the largest signatures run to tens of MB. The row +-- is written once and never updated. +CREATE TABLE IF NOT EXISTS target_signature ( + queue VARCHAR(255) NOT NULL, + batch_id VARCHAR(255) NOT NULL, + base_revision VARCHAR(255) NOT NULL, + targets LONGBLOB NOT NULL, + created_at BIGINT NOT NULL, + PRIMARY KEY (queue, batch_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/submitqueue/extension/conflict/tango/mysql/signature_codec.go b/submitqueue/extension/conflict/tango/mysql/signature_codec.go new file mode 100644 index 000000000..0a216badc --- /dev/null +++ b/submitqueue/extension/conflict/tango/mysql/signature_codec.go @@ -0,0 +1,64 @@ +// 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 ( + "encoding/binary" + "fmt" + "math" + + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango" +) + +// idBytes is the width of an encoded target ID. +const idBytes = 8 + +// encodeSignatureTargets packs targets as a flat byte string: per target, the ID +// as 8 big-endian bytes followed by the distance as a signed varint. The result +// is never nil, because a nil slice would be written as NULL into a NOT NULL +// column. +func encodeSignatureTargets(targets []tango.SignatureTarget) []byte { + encoded := make([]byte, 0, len(targets)*(idBytes+1)) + for _, target := range targets { + encoded = binary.BigEndian.AppendUint64(encoded, target.ID) + encoded = binary.AppendVarint(encoded, int64(target.Distance)) + } + return encoded +} + +// decodeSignatureTargets reverses encodeSignatureTargets. It returns an error on +// truncated input or a distance outside the int32 range. +func decodeSignatureTargets(encoded []byte) ([]tango.SignatureTarget, error) { + var targets []tango.SignatureTarget + for offset := 0; offset < len(encoded); { + if len(encoded)-offset < idBytes { + return nil, fmt.Errorf("truncated target ID at offset %d of %d bytes", offset, len(encoded)) + } + id := binary.BigEndian.Uint64(encoded[offset:]) + offset += idBytes + + distance, read := binary.Varint(encoded[offset:]) + if read <= 0 { + return nil, fmt.Errorf("invalid distance varint at offset %d of %d bytes", offset, len(encoded)) + } + offset += read + if distance < math.MinInt32 || distance > math.MaxInt32 { + return nil, fmt.Errorf("distance %d for target ID %d is outside the int32 range", distance, id) + } + + targets = append(targets, tango.SignatureTarget{ID: id, Distance: int32(distance)}) + } + return targets, nil +} diff --git a/submitqueue/extension/conflict/tango/mysql/signature_codec_test.go b/submitqueue/extension/conflict/tango/mysql/signature_codec_test.go new file mode 100644 index 000000000..b16229368 --- /dev/null +++ b/submitqueue/extension/conflict/tango/mysql/signature_codec_test.go @@ -0,0 +1,116 @@ +// 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 ( + "encoding/binary" + "math" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango" +) + +func TestSignatureTargetsCodec_RoundTrip(t *testing.T) { + tests := []struct { + name string + targets []tango.SignatureTarget + }{ + { + name: "empty", + targets: nil, + }, + { + name: "single target", + targets: []tango.SignatureTarget{{ID: 1, Distance: 0}}, + }, + { + name: "several targets in ascending ID order", + targets: []tango.SignatureTarget{ + {ID: 0x735cc8dafbb93f02, Distance: 0}, + {ID: 0x90afc9166e08f639, Distance: 1}, + {ID: 0xfcf7cdab318e139e, Distance: 2}, + }, + }, + { + name: "extreme IDs and distances", + targets: []tango.SignatureTarget{ + {ID: 0, Distance: math.MaxInt32}, + {ID: math.MaxUint64, Distance: 63}, + {ID: math.MaxUint64 - 1, Distance: 64}, + }, + }, + { + name: "negative distance", + targets: []tango.SignatureTarget{{ID: 7, Distance: -1}}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := decodeSignatureTargets(encodeSignatureTargets(tt.targets)) + require.NoError(t, err) + assert.Equal(t, tt.targets, got) + }) + } +} + +func TestEncodeSignatureTargets_EmptyIsNotNil(t *testing.T) { + assert.NotNil(t, encodeSignatureTargets(nil)) + assert.Empty(t, encodeSignatureTargets(nil)) +} + +func TestEncodeSignatureTargets_Layout(t *testing.T) { + got := encodeSignatureTargets([]tango.SignatureTarget{{ID: 0x0102030405060708, Distance: 2}}) + + assert.Equal(t, []byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x04}, got) +} + +func TestDecodeSignatureTargets_Invalid(t *testing.T) { + idOnly := binary.BigEndian.AppendUint64(nil, 1) + + tests := []struct { + name string + encoded []byte + }{ + { + name: "truncated ID", + encoded: []byte{0x01, 0x02, 0x03}, + }, + { + name: "ID without a distance", + encoded: idOnly, + }, + { + name: "unterminated distance varint", + encoded: append(append([]byte{}, idOnly...), 0x80), + }, + { + name: "distance outside int32", + encoded: binary.AppendVarint(append([]byte{}, idOnly...), math.MaxInt32+1), + }, + { + name: "valid target followed by a truncated one", + encoded: append(binary.AppendVarint(append([]byte{}, idOnly...), 0), 0x01, 0x02), + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := decodeSignatureTargets(tt.encoded) + assert.Error(t, err) + }) + } +} diff --git a/submitqueue/extension/conflict/tango/mysql/signature_store.go b/submitqueue/extension/conflict/tango/mysql/signature_store.go new file mode 100644 index 000000000..b10524e1e --- /dev/null +++ b/submitqueue/extension/conflict/tango/mysql/signature_store.go @@ -0,0 +1,91 @@ +// 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 signatureStore 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 +} + +// NewSignatureStore creates a new MySQL-backed TargetSignatureStore. +func NewSignatureStore(db *sql.DB, scope tally.Scope, queue string) tango.SignatureStore { + return &signatureStore{db: db, scope: scope, queue: queue} +} + +// Create inserts a signature. Returns ErrAlreadyExists if the batch already has one. +func (s *signatureStore) Create(ctx context.Context, signature tango.TargetSignature) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if signature.Queue != s.queue { + return fmt.Errorf("target signature batch_id=%s queue %q does not match the store's bound queue %q", signature.BatchID, signature.Queue, s.queue) + } + + _, err := s.db.ExecContext(ctx, + "INSERT INTO target_signature (queue, batch_id, base_revision, targets, created_at) VALUES (?, ?, ?, ?, ?)", + signature.Queue, signature.BatchID, signature.BaseRevision, encodeSignatureTargets(signature.Targets), signature.CreatedAt, + ) + if err != nil { + var mysqlErr *mysql.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry { + return fmt.Errorf("target signature batch_id=%s: %w", signature.BatchID, storage.ErrAlreadyExists) + } + return fmt.Errorf("failed to insert target signature batch_id=%s: %w", signature.BatchID, err) + } + return nil +} + +// Get returns the bound queue's signature for batchID. The bool is false when +// the batch has none. +func (s *signatureStore) Get(ctx context.Context, batchID string) (ret tango.TargetSignature, found bool, retErr error) { + op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + signature := tango.TargetSignature{Queue: s.queue, BatchID: batchID} + var encodedTargets []byte + err := s.db.QueryRowContext(ctx, + "SELECT base_revision, targets, created_at FROM target_signature WHERE queue = ? AND batch_id = ?", + s.queue, batchID, + ).Scan(&signature.BaseRevision, &encodedTargets, &signature.CreatedAt) + if errors.Is(err, sql.ErrNoRows) { + return tango.TargetSignature{}, false, nil + } + if err != nil { + return tango.TargetSignature{}, false, fmt.Errorf("failed to get target signature batch_id=%s from the database: %w", batchID, err) + } + + signature.Targets, err = decodeSignatureTargets(encodedTargets) + if err != nil { + return tango.TargetSignature{}, false, fmt.Errorf("failed to decode targets of target signature batch_id=%s: %w", batchID, err) + } + return signature, true, nil +} diff --git a/submitqueue/extension/conflict/tango/mysql/signature_store_test.go b/submitqueue/extension/conflict/tango/mysql/signature_store_test.go new file mode 100644 index 000000000..0f407ffa9 --- /dev/null +++ b/submitqueue/extension/conflict/tango/mysql/signature_store_test.go @@ -0,0 +1,223 @@ +// 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-go/tally" + + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +func testMetrics() tally.Scope { + return tally.NoopScope +} + +func setupSignatureStoreTest(t *testing.T) (*sql.DB, sqlmock.Sqlmock, tango.SignatureStore) { + t.Helper() + + db, mock, err := sqlmock.New() + require.NoError(t, err) + + store := NewSignatureStore(db, testMetrics(), "monorepo") + + return db, mock, store +} + +func testTargetSignature() tango.TargetSignature { + return tango.TargetSignature{ + Queue: "monorepo", + BatchID: "1", + BaseRevision: "9f2c1d0e3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d", + Targets: []tango.SignatureTarget{ + {ID: 10, Distance: 0}, + {ID: 20, Distance: 1}, + }, + CreatedAt: 1700000000000, + } +} + +func TestTargetSignatureStore_Create(t *testing.T) { + signature := testTargetSignature() + mismatched := signature + mismatched.Queue = "other-queue" + + tests := []struct { + name string + signature tango.TargetSignature + setup func(mock sqlmock.Sqlmock) + wantErr bool + wantErrIs error + }{ + { + name: "success", + signature: signature, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO target_signature"). + WithArgs("monorepo", "1", signature.BaseRevision, encodeSignatureTargets(signature.Targets), signature.CreatedAt). + WillReturnResult(sqlmock.NewResult(1, 1)) + }, + }, + { + name: "empty signature is written as an empty value", + signature: tango.TargetSignature{Queue: "monorepo", BatchID: "2", BaseRevision: signature.BaseRevision, CreatedAt: 5}, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO target_signature"). + WithArgs("monorepo", "2", signature.BaseRevision, []byte{}, int64(5)). + WillReturnResult(sqlmock.NewResult(1, 1)) + }, + }, + { + name: "duplicate batch", + signature: signature, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO target_signature"). + WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) + }, + wantErr: true, + wantErrIs: storage.ErrAlreadyExists, + }, + { + name: "insert error", + signature: signature, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO target_signature"). + WillReturnError(fmt.Errorf("connection reset")) + }, + wantErr: true, + }, + { + name: "queue differs from the bound queue", + signature: mismatched, + setup: func(mock sqlmock.Sqlmock) {}, + wantErr: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, store := setupSignatureStoreTest(t) + defer db.Close() + tt.setup(mock) + + err := store.Create(context.Background(), tt.signature) + + 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 TestTargetSignatureStore_Get(t *testing.T) { + want := testTargetSignature() + columns := []string{"base_revision", "targets", "created_at"} + + tests := []struct { + name string + setup func(mock sqlmock.Sqlmock) + want tango.TargetSignature + wantFound bool + wantErr bool + }{ + { + name: "found", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT base_revision, targets, created_at FROM target_signature"). + WithArgs("monorepo", want.BatchID). + WillReturnRows(sqlmock.NewRows(columns). + AddRow(want.BaseRevision, encodeSignatureTargets(want.Targets), want.CreatedAt)) + }, + want: want, + wantFound: true, + }, + { + name: "found with no targets", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT base_revision, targets, created_at FROM target_signature"). + WithArgs("monorepo", want.BatchID). + WillReturnRows(sqlmock.NewRows(columns). + AddRow(want.BaseRevision, []byte{}, want.CreatedAt)) + }, + want: tango.TargetSignature{ + Queue: want.Queue, + BatchID: want.BatchID, + BaseRevision: want.BaseRevision, + CreatedAt: want.CreatedAt, + }, + wantFound: true, + }, + { + name: "not found", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT base_revision, targets, created_at FROM target_signature"). + WithArgs("monorepo", want.BatchID). + WillReturnError(sql.ErrNoRows) + }, + }, + { + name: "query error", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT base_revision, targets, created_at FROM target_signature"). + WithArgs("monorepo", want.BatchID). + WillReturnError(fmt.Errorf("connection reset")) + }, + wantErr: true, + }, + { + name: "stored targets cannot be decoded", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT base_revision, targets, created_at FROM target_signature"). + WithArgs("monorepo", want.BatchID). + WillReturnRows(sqlmock.NewRows(columns). + AddRow(want.BaseRevision, []byte{0x01, 0x02, 0x03}, want.CreatedAt)) + }, + wantErr: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, store := setupSignatureStoreTest(t) + defer db.Close() + tt.setup(mock) + + got, found, err := store.Get(context.Background(), want.BatchID) + + if tt.wantErr { + require.Error(t, err) + assert.False(t, found) + } 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/test/integration/submitqueue/extension/conflict/tango/mysql/BUILD.bazel b/test/integration/submitqueue/extension/conflict/tango/mysql/BUILD.bazel new file mode 100644 index 000000000..c6ed1fbf0 --- /dev/null +++ b/test/integration/submitqueue/extension/conflict/tango/mysql/BUILD.bazel @@ -0,0 +1,25 @@ +load("@rules_go//go:def.bzl", "go_test") + +go_test( + name = "go_default_test", + srcs = ["signature_store_test.go"], + data = [ + "docker-compose.yml", + "//submitqueue/extension/conflict/tango/mysql/schema", + ], + tags = [ + "integration", + "requires-network", + ], + deps = [ + "//submitqueue/extension/conflict/tango:go_default_library", + "//submitqueue/extension/conflict/tango/mysql:go_default_library", + "//submitqueue/extension/storage:go_default_library", + "//test/testutil:go_default_library", + "@com_github_go_sql_driver_mysql//:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@com_github_stretchr_testify//suite:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + ], +) diff --git a/test/integration/submitqueue/extension/conflict/tango/mysql/docker-compose.yml b/test/integration/submitqueue/extension/conflict/tango/mysql/docker-compose.yml new file mode 100644 index 000000000..54b533d2b --- /dev/null +++ b/test/integration/submitqueue/extension/conflict/tango/mysql/docker-compose.yml @@ -0,0 +1,24 @@ +# Docker Compose for the tango analyzer's MySQL store tests +# Tests the MySQL implementation in isolation + +services: + # MySQL database for the signature store + mysql: + image: mysql:8.0 + environment: + MYSQL_ROOT_PASSWORD: root + MYSQL_DATABASE: submitqueue + MYSQL_INITDB_SKIP_TZINFO: ${SQ_MYSQL_INITDB_SKIP_TZINFO:-} + ports: + - "3306" # Random ephemeral port to avoid conflicts + volumes: + - type: ${SQ_MYSQL_DATA_MOUNT_TYPE:-volume} + target: /var/lib/mysql + healthcheck: + # Use 127.0.0.1 (TCP) instead of localhost (Unix socket). MySQL treats + # "localhost" as a socket connection, which can be ready before the TCP + # listener — causing dependent services that connect over TCP to fail. + test: ["CMD", "mysqladmin", "ping", "-h", "127.0.0.1", "-proot"] + interval: 1s + timeout: 2s + retries: 60 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 new file mode 100644 index 000000000..9f47e8637 --- /dev/null +++ b/test/integration/submitqueue/extension/conflict/tango/mysql/signature_store_test.go @@ -0,0 +1,132 @@ +// 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/go-sql-driver/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/stretchr/testify/suite" + "github.com/uber-go/tally" + + "github.com/uber/submitqueue/submitqueue/extension/conflict/tango" + tangomysql "github.com/uber/submitqueue/submitqueue/extension/conflict/tango/mysql" + "github.com/uber/submitqueue/submitqueue/extension/storage" + "github.com/uber/submitqueue/test/testutil" +) + +// SignatureStoreIntegrationSuite runs the MySQL signature store against a real +// database. +type SignatureStoreIntegrationSuite struct { + suite.Suite + stack *testutil.ComposeStack + db *sql.DB + log *testutil.TestLogger +} + +func TestSignatureStoreIntegration(t *testing.T) { + suite.Run(t, new(SignatureStoreIntegrationSuite)) +} + +func (s *SignatureStoreIntegrationSuite) SetupSuite() { + t := s.T() + ctx := context.Background() + s.log = testutil.NewTestLogger(t) + + s.stack = testutil.NewComposeStack(t, s.log, ctx, "docker-compose.yml", "ext-submitqueue-conflict-tango-mysql") + require.NoError(t, s.stack.Up(), "failed to start compose stack") + + var err error + s.db, err = s.stack.ConnectMySQLService("mysql") + require.NoError(t, err, "failed to connect to MySQL") + testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("submitqueue/extension/conflict/tango/mysql/schema")) + + t.Cleanup(func() { s.db.Close() }) +} + +func (s *SignatureStoreIntegrationSuite) forQueue(queue string) tango.SignatureStore { + return tangomysql.NewSignatureStore(s.db, tally.NoopScope, queue) +} + +func (s *SignatureStoreIntegrationSuite) TestSignatureStore() { + t := s.T() + ctx := context.Background() + storeA := s.forQueue("sig-queue-a") + storeB := s.forQueue("sig-queue-b") + + // 100k entries encode to roughly 900KB, beyond a BLOB column's 64KB. + large := make([]tango.SignatureTarget, 100000) + for i := range large { + large[i] = tango.SignatureTarget{ID: uint64(i) * 3, Distance: int32(i % 5)} + } + + tests := []struct { + name string + signature tango.TargetSignature + }{ + { + name: "several targets", + signature: tango.TargetSignature{ + Queue: "sig-queue-a", BatchID: "small", BaseRevision: "rev-1", CreatedAt: 1700000000000, + Targets: []tango.SignatureTarget{{ID: 1, Distance: 0}, {ID: 1 << 63, Distance: 2}, {ID: ^uint64(0), Distance: 7}}, + }, + }, + { + name: "empty signature", + signature: tango.TargetSignature{Queue: "sig-queue-a", BatchID: "empty", BaseRevision: "rev-1", CreatedAt: 1}, + }, + { + name: "large signature", + signature: tango.TargetSignature{Queue: "sig-queue-a", BatchID: "large", BaseRevision: "rev-2", CreatedAt: 2, Targets: large}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.NoError(t, storeA.Create(ctx, tt.signature)) + + got, found, err := storeA.Get(ctx, tt.signature.BatchID) + require.NoError(t, err) + require.True(t, found) + assert.Equal(t, tt.signature, got) + }) + } + + t.Run("missing batch is not found", func(t *testing.T) { + _, found, err := storeA.Get(ctx, "missing") + require.NoError(t, err) + assert.False(t, found) + }) + + t.Run("second create keeps the first signature", func(t *testing.T) { + duplicate := tango.TargetSignature{Queue: "sig-queue-a", BatchID: "small", BaseRevision: "rev-other", CreatedAt: 9} + require.ErrorIs(t, storeA.Create(ctx, duplicate), storage.ErrAlreadyExists) + + got, found, err := storeA.Get(ctx, "small") + require.NoError(t, err) + require.True(t, found) + assert.Equal(t, "rev-1", got.BaseRevision) + }) + + t.Run("invisible through another queue", func(t *testing.T) { + _, found, err := storeB.Get(ctx, "small") + require.NoError(t, err) + assert.False(t, found) + require.Error(t, storeB.Create(ctx, tango.TargetSignature{Queue: "sig-queue-a", BatchID: "mismatch"}), "a mismatched-queue write must be rejected") + }) +} diff --git a/tool/linter/queueshard/main.go b/tool/linter/queueshard/main.go index ef7509548..ec8e03d96 100644 --- a/tool/linter/queueshard/main.go +++ b/tool/linter/queueshard/main.go @@ -36,6 +36,7 @@ import ( var schemaShardColumns = map[string]map[string]bool{ "submitqueue/orchestrator/extension/storage/mysql/schema": {"queue": true, "name": true}, "submitqueue/gateway/extension/storage/mysql/schema": {"queue": true, "name": true}, + "submitqueue/extension/conflict/tango/mysql/schema": {"queue": true, "name": true}, "stovepipe/extension/storage/mysql/schema": {"queue": true, "name": true}, "platform/extension/counter/mysql/schema": {"queue": true, "name": true}, "platform/extension/messagequeue/mysql/schema": {"tenant": true}, @@ -45,6 +46,7 @@ var schemaShardColumns = map[string]map[string]bool{ var schemaRoots = []string{ "submitqueue/orchestrator/extension/storage/mysql/schema", "submitqueue/gateway/extension/storage/mysql/schema", + "submitqueue/extension/conflict/tango/mysql/schema", "stovepipe/extension/storage/mysql/schema", "platform/extension/counter/mysql/schema", "platform/extension/messagequeue/mysql/schema", From 2d8623f58529e20f57d614eb06f048f5190be343 Mon Sep 17 00:00:00 2001 From: manjari Date: Fri, 16 Oct 2026 17:14:53 +0000 Subject: [PATCH 2/2] docs(conflict): Document signature target ordering in the tango mysql codec --- submitqueue/extension/conflict/tango/mysql/mysql.go | 5 +++++ .../extension/conflict/tango/mysql/signature_codec.go | 7 +++---- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/submitqueue/extension/conflict/tango/mysql/mysql.go b/submitqueue/extension/conflict/tango/mysql/mysql.go index 4a0d0ddad..9511efe5c 100644 --- a/submitqueue/extension/conflict/tango/mysql/mysql.go +++ b/submitqueue/extension/conflict/tango/mysql/mysql.go @@ -13,6 +13,11 @@ // limitations under the License. // Package mysql implements the tango analyzer's stores over MySQL. +// +// Targets are stored in the order given. TargetSignature.Targets is sorted by ID +// with no duplicates, established when the signature is built. The store neither +// sorts nor validates. The order is not needed for correctness; it lets +// signatures be intersected without building a set. package mysql // mysqlErrDuplicateEntry is MySQL error code 1062 ("Duplicate entry"), returned on a unique or primary key violation. diff --git a/submitqueue/extension/conflict/tango/mysql/signature_codec.go b/submitqueue/extension/conflict/tango/mysql/signature_codec.go index 0a216badc..b23dc50b7 100644 --- a/submitqueue/extension/conflict/tango/mysql/signature_codec.go +++ b/submitqueue/extension/conflict/tango/mysql/signature_codec.go @@ -25,10 +25,9 @@ import ( // idBytes is the width of an encoded target ID. const idBytes = 8 -// encodeSignatureTargets packs targets as a flat byte string: per target, the ID -// as 8 big-endian bytes followed by the distance as a signed varint. The result -// is never nil, because a nil slice would be written as NULL into a NOT NULL -// column. +// encodeSignatureTargets packs each target as its ID (8 big-endian bytes) and a +// signed varint distance, in the order given. The result is never nil, because a +// nil slice would be written as NULL into a NOT NULL column. func encodeSignatureTargets(targets []tango.SignatureTarget) []byte { encoded := make([]byte, 0, len(targets)*(idBytes+1)) for _, target := range targets {