Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions submitqueue/extension/conflict/tango/mysql/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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",
],
Expand Down
85 changes: 85 additions & 0 deletions submitqueue/extension/conflict/tango/mysql/dictionary_store.go
Original file line number Diff line number Diff line change
@@ -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
}
164 changes: 164 additions & 0 deletions submitqueue/extension/conflict/tango/mysql/dictionary_store_test.go
Original file line number Diff line number Diff line change
@@ -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())
})
}
}
Original file line number Diff line number Diff line change
@@ -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;
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
Original file line number Diff line number Diff line change
@@ -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")
})
}
Loading
Loading