Skip to content

Commit 29d5822

Browse files
manjari25manjari25
committed
refactor(conflict): Move target dictionary store to tango mysql
1 parent fa0cf0f commit 29d5822

7 files changed

Lines changed: 370 additions & 8 deletions

File tree

‎submitqueue/extension/conflict/tango/mysql/BUILD.bazel‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
33
go_library(
44
name = "go_default_library",
55
srcs = [
6+
"dictionary_store.go",
67
"mysql.go",
78
"signature_codec.go",
89
"signature_store.go",
@@ -25,6 +26,7 @@ go_library(
2526
go_test(
2627
name = "go_default_test",
2728
srcs = [
29+
"dictionary_store_test.go",
2830
"signature_codec_test.go",
2931
"signature_store_test.go",
3032
],

‎submitqueue/extension/conflict/tango/mysql/dictionary_store.go‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package mysql
16+
17+
import (
18+
"context"
19+
"database/sql"
20+
"errors"
21+
"fmt"
22+
23+
"github.com/go-sql-driver/mysql"
24+
"github.com/uber-go/tally"
25+
26+
"github.com/uber/submitqueue/platform/metrics"
27+
"github.com/uber/submitqueue/submitqueue/extension/conflict/tango"
28+
"github.com/uber/submitqueue/submitqueue/extension/storage"
29+
)
30+
31+
type dictionaryStore struct {
32+
db *sql.DB
33+
scope tally.Scope
34+
// queue is the queue name this store instance is bound to; every read and
35+
// write is scoped to it.
36+
queue string
37+
}
38+
39+
// NewDictionaryStore creates a new MySQL-backed TargetDictionaryStore.
40+
func NewDictionaryStore(db *sql.DB, scope tally.Scope, queue string) tango.DictionaryStore {
41+
return &dictionaryStore{db: db, scope: scope, queue: queue}
42+
}
43+
44+
// Create inserts a label. Returns ErrAlreadyExists if the ID is already recorded.
45+
func (s *dictionaryStore) Create(ctx context.Context, label tango.TargetLabel) (retErr error) {
46+
op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets)
47+
defer func() { op.Complete(retErr) }()
48+
49+
if label.Queue != s.queue {
50+
return fmt.Errorf("target label id=%d queue %q does not match the store's bound queue %q", label.ID, label.Queue, s.queue)
51+
}
52+
53+
_, err := s.db.ExecContext(ctx,
54+
"INSERT INTO target_dictionary (queue, id, label, created_at) VALUES (?, ?, ?, ?)",
55+
label.Queue, label.ID, label.Label, label.CreatedAt,
56+
)
57+
if err != nil {
58+
var mysqlErr *mysql.MySQLError
59+
if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry {
60+
return fmt.Errorf("target label id=%d: %w", label.ID, storage.ErrAlreadyExists)
61+
}
62+
return fmt.Errorf("failed to insert target label id=%d: %w", label.ID, err)
63+
}
64+
return nil
65+
}
66+
67+
// Get returns the bound queue's label for id. The bool is false when none is
68+
// recorded.
69+
func (s *dictionaryStore) Get(ctx context.Context, id uint64) (ret tango.TargetLabel, found bool, retErr error) {
70+
op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
71+
defer func() { op.Complete(retErr) }()
72+
73+
label := tango.TargetLabel{Queue: s.queue, ID: id}
74+
err := s.db.QueryRowContext(ctx,
75+
"SELECT label, created_at FROM target_dictionary WHERE queue = ? AND id = ?",
76+
s.queue, id,
77+
).Scan(&label.Label, &label.CreatedAt)
78+
if errors.Is(err, sql.ErrNoRows) {
79+
return tango.TargetLabel{}, false, nil
80+
}
81+
if err != nil {
82+
return tango.TargetLabel{}, false, fmt.Errorf("failed to get target label id=%d from the database: %w", id, err)
83+
}
84+
return label, true, nil
85+
}

‎submitqueue/extension/conflict/tango/mysql/dictionary_store_test.go‎

Lines changed: 164 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,164 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package mysql
16+
17+
import (
18+
"context"
19+
"database/sql"
20+
"fmt"
21+
"testing"
22+
23+
"github.com/DATA-DOG/go-sqlmock"
24+
"github.com/go-sql-driver/mysql"
25+
"github.com/stretchr/testify/assert"
26+
"github.com/stretchr/testify/require"
27+
28+
"github.com/uber/submitqueue/submitqueue/extension/conflict/tango"
29+
"github.com/uber/submitqueue/submitqueue/extension/storage"
30+
)
31+
32+
func setupDictionaryStoreTest(t *testing.T) (*sql.DB, sqlmock.Sqlmock, tango.DictionaryStore) {
33+
t.Helper()
34+
35+
db, mock, err := sqlmock.New()
36+
require.NoError(t, err)
37+
38+
return db, mock, NewDictionaryStore(db, testMetrics(), "monorepo")
39+
}
40+
41+
func TestTargetDictionaryStore_Create(t *testing.T) {
42+
label := tango.TargetLabel{Queue: "monorepo", ID: 42, Label: "//app/server:lib", CreatedAt: 1700000000000}
43+
mismatched := label
44+
mismatched.Queue = "other-queue"
45+
46+
tests := []struct {
47+
name string
48+
label tango.TargetLabel
49+
setup func(mock sqlmock.Sqlmock)
50+
wantErr bool
51+
wantErrIs error
52+
}{
53+
{
54+
name: "success",
55+
label: label,
56+
setup: func(mock sqlmock.Sqlmock) {
57+
mock.ExpectExec("INSERT INTO target_dictionary").
58+
WithArgs("monorepo", uint64(42), label.Label, label.CreatedAt).
59+
WillReturnResult(sqlmock.NewResult(1, 1))
60+
},
61+
},
62+
{
63+
name: "id already recorded",
64+
label: label,
65+
setup: func(mock sqlmock.Sqlmock) {
66+
mock.ExpectExec("INSERT INTO target_dictionary").
67+
WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry})
68+
},
69+
wantErr: true,
70+
wantErrIs: storage.ErrAlreadyExists,
71+
},
72+
{
73+
name: "insert error",
74+
label: label,
75+
setup: func(mock sqlmock.Sqlmock) {
76+
mock.ExpectExec("INSERT INTO target_dictionary").
77+
WillReturnError(fmt.Errorf("data too long for column label"))
78+
},
79+
wantErr: true,
80+
},
81+
{
82+
name: "queue differs from the bound queue",
83+
label: mismatched,
84+
setup: func(mock sqlmock.Sqlmock) {},
85+
wantErr: true,
86+
},
87+
}
88+
for _, tt := range tests {
89+
t.Run(tt.name, func(t *testing.T) {
90+
db, mock, store := setupDictionaryStoreTest(t)
91+
defer db.Close()
92+
tt.setup(mock)
93+
94+
err := store.Create(context.Background(), tt.label)
95+
96+
if tt.wantErr {
97+
require.Error(t, err)
98+
if tt.wantErrIs != nil {
99+
assert.ErrorIs(t, err, tt.wantErrIs)
100+
}
101+
} else {
102+
require.NoError(t, err)
103+
}
104+
assert.NoError(t, mock.ExpectationsWereMet())
105+
})
106+
}
107+
}
108+
109+
func TestTargetDictionaryStore_Get(t *testing.T) {
110+
columns := []string{"label", "created_at"}
111+
112+
tests := []struct {
113+
name string
114+
setup func(mock sqlmock.Sqlmock)
115+
want tango.TargetLabel
116+
wantFound bool
117+
wantErr bool
118+
}{
119+
{
120+
name: "found",
121+
setup: func(mock sqlmock.Sqlmock) {
122+
mock.ExpectQuery("SELECT label, created_at FROM target_dictionary").
123+
WithArgs("monorepo", uint64(42)).
124+
WillReturnRows(sqlmock.NewRows(columns).AddRow("//app:lib", int64(10)))
125+
},
126+
want: tango.TargetLabel{Queue: "monorepo", ID: 42, Label: "//app:lib", CreatedAt: 10},
127+
wantFound: true,
128+
},
129+
{
130+
name: "none recorded",
131+
setup: func(mock sqlmock.Sqlmock) {
132+
mock.ExpectQuery("SELECT label, created_at FROM target_dictionary").
133+
WithArgs("monorepo", uint64(42)).
134+
WillReturnRows(sqlmock.NewRows(columns))
135+
},
136+
},
137+
{
138+
name: "query error",
139+
setup: func(mock sqlmock.Sqlmock) {
140+
mock.ExpectQuery("SELECT label, created_at FROM target_dictionary").
141+
WillReturnError(fmt.Errorf("connection reset"))
142+
},
143+
wantErr: true,
144+
},
145+
}
146+
for _, tt := range tests {
147+
t.Run(tt.name, func(t *testing.T) {
148+
db, mock, store := setupDictionaryStoreTest(t)
149+
defer db.Close()
150+
tt.setup(mock)
151+
152+
got, found, err := store.Get(context.Background(), 42)
153+
154+
if tt.wantErr {
155+
require.Error(t, err)
156+
} else {
157+
require.NoError(t, err)
158+
assert.Equal(t, tt.wantFound, found)
159+
assert.Equal(t, tt.want, got)
160+
}
161+
assert.NoError(t, mock.ExpectationsWereMet())
162+
})
163+
}
164+
}

‎submitqueue/extension/conflict/tango/mysql/schema/target_dictionary.sql‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
-- Rows are written once and never updated. Two labels can share an id on a
2+
-- hash collision; the first one written stays.
3+
CREATE TABLE IF NOT EXISTS target_dictionary (
4+
queue VARCHAR(255) NOT NULL,
5+
id BIGINT UNSIGNED NOT NULL,
6+
label TEXT NOT NULL,
7+
created_at BIGINT NOT NULL,
8+
PRIMARY KEY (queue, id)
9+
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

‎test/integration/submitqueue/extension/conflict/tango/mysql/BUILD.bazel‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_test")
22

33
go_test(
44
name = "go_default_test",
5-
srcs = ["signature_store_test.go"],
5+
srcs = [
6+
"dictionary_store_test.go",
7+
"signature_store_test.go",
8+
],
69
data = [
710
"docker-compose.yml",
811
"//submitqueue/extension/conflict/tango/mysql/schema",

‎test/integration/submitqueue/extension/conflict/tango/mysql/dictionary_store_test.go‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package mysql
16+
17+
import (
18+
"context"
19+
"strings"
20+
"testing"
21+
22+
"github.com/stretchr/testify/assert"
23+
"github.com/stretchr/testify/require"
24+
25+
"github.com/uber/submitqueue/submitqueue/extension/conflict/tango"
26+
"github.com/uber/submitqueue/submitqueue/extension/storage"
27+
)
28+
29+
func (s *StoreIntegrationSuite) TestDictionaryStore() {
30+
t := s.T()
31+
ctx := context.Background()
32+
storeA := s.forQueueDictionary("dict-queue-a")
33+
storeB := s.forQueueDictionary("dict-queue-b")
34+
35+
// An ID with the high bit set must survive the unsigned column round trip.
36+
const highID = uint64(1) << 63
37+
label := func(id uint64, name string) tango.TargetLabel {
38+
return tango.TargetLabel{Queue: "dict-queue-a", ID: id, Label: name, CreatedAt: 1700000000000}
39+
}
40+
41+
t.Run("round trip", func(t *testing.T) {
42+
require.NoError(t, storeA.Create(ctx, label(7, "//app/server:lib")))
43+
44+
got, found, err := storeA.Get(ctx, 7)
45+
require.NoError(t, err)
46+
assert.True(t, found)
47+
assert.Equal(t, label(7, "//app/server:lib"), got)
48+
})
49+
50+
t.Run("ID with the high bit set", func(t *testing.T) {
51+
require.NoError(t, storeA.Create(ctx, label(highID, "//high:bit")))
52+
53+
got, found, err := storeA.Get(ctx, highID)
54+
require.NoError(t, err)
55+
assert.True(t, found)
56+
assert.Equal(t, label(highID, "//high:bit"), got)
57+
})
58+
59+
t.Run("long label", func(t *testing.T) {
60+
long := "//" + strings.Repeat("p", 900) + ":target"
61+
require.NoError(t, storeA.Create(ctx, label(11, long)))
62+
63+
got, found, err := storeA.Get(ctx, 11)
64+
require.NoError(t, err)
65+
assert.True(t, found)
66+
assert.Equal(t, label(11, long), got)
67+
})
68+
69+
t.Run("a second label for a recorded ID is rejected and the first stays", func(t *testing.T) {
70+
require.NoError(t, storeA.Create(ctx, label(9, "//a:x")))
71+
require.ErrorIs(t, storeA.Create(ctx, label(9, "//b:y")), storage.ErrAlreadyExists)
72+
73+
got, found, err := storeA.Get(ctx, 9)
74+
require.NoError(t, err)
75+
assert.True(t, found)
76+
assert.Equal(t, label(9, "//a:x"), got)
77+
})
78+
79+
t.Run("repeated create returns ErrAlreadyExists", func(t *testing.T) {
80+
require.ErrorIs(t, storeA.Create(ctx, label(7, "//app/server:lib")), storage.ErrAlreadyExists)
81+
})
82+
83+
t.Run("unknown ID is not found", func(t *testing.T) {
84+
_, found, err := storeA.Get(ctx, 12345)
85+
require.NoError(t, err)
86+
assert.False(t, found)
87+
})
88+
89+
t.Run("invisible through another queue", func(t *testing.T) {
90+
_, found, err := storeB.Get(ctx, 7)
91+
require.NoError(t, err)
92+
assert.False(t, found)
93+
require.Error(t, storeB.Create(ctx, label(7, "//app/server:lib")), "a mismatched-queue write must be rejected")
94+
})
95+
}

0 commit comments

Comments
 (0)