Skip to content

Commit fa0cf0f

Browse files
manjari25manjari25
committed
refactor(conflict): Move target signature store to tango mysql
1 parent 842f382 commit fa0cf0f

12 files changed

Lines changed: 753 additions & 0 deletions

File tree

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

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
load("@rules_go//go:def.bzl", "go_library", "go_test")
2+
3+
go_library(
4+
name = "go_default_library",
5+
srcs = [
6+
"mysql.go",
7+
"signature_codec.go",
8+
"signature_store.go",
9+
],
10+
importpath = "github.com/uber/submitqueue/submitqueue/extension/conflict/tango/mysql",
11+
visibility = [
12+
"//service/submitqueue/orchestrator:__subpackages__",
13+
"//submitqueue/extension/conflict/tango:__subpackages__",
14+
"//test/integration/submitqueue/extension/conflict/tango:__subpackages__",
15+
],
16+
deps = [
17+
"//platform/metrics:go_default_library",
18+
"//submitqueue/extension/conflict/tango:go_default_library",
19+
"//submitqueue/extension/storage:go_default_library",
20+
"@com_github_go_sql_driver_mysql//:go_default_library",
21+
"@com_github_uber_go_tally//:go_default_library",
22+
],
23+
)
24+
25+
go_test(
26+
name = "go_default_test",
27+
srcs = [
28+
"signature_codec_test.go",
29+
"signature_store_test.go",
30+
],
31+
embed = [":go_default_library"],
32+
deps = [
33+
"//submitqueue/extension/conflict/tango:go_default_library",
34+
"//submitqueue/extension/storage:go_default_library",
35+
"@com_github_data_dog_go_sqlmock//:go_default_library",
36+
"@com_github_go_sql_driver_mysql//:go_default_library",
37+
"@com_github_stretchr_testify//assert:go_default_library",
38+
"@com_github_stretchr_testify//require:go_default_library",
39+
"@com_github_uber_go_tally//:go_default_library",
40+
],
41+
)

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

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
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 implements the tango analyzer's stores over MySQL.
16+
package mysql
17+
18+
// mysqlErrDuplicateEntry is MySQL error code 1062 ("Duplicate entry"), returned on a unique or primary key violation.
19+
const mysqlErrDuplicateEntry = 1062

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
filegroup(
2+
name = "schema",
3+
srcs = glob(["*.sql"]),
4+
visibility = ["//visibility:public"],
5+
)

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

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
-- targets is the binary encoding of the signature's sorted (id, distance)
2+
-- entries. LONGBLOB because the largest signatures run to tens of MB. The row
3+
-- is written once and never updated.
4+
CREATE TABLE IF NOT EXISTS target_signature (
5+
queue VARCHAR(255) NOT NULL,
6+
batch_id VARCHAR(255) NOT NULL,
7+
base_revision VARCHAR(255) NOT NULL,
8+
targets LONGBLOB NOT NULL,
9+
created_at BIGINT NOT NULL,
10+
PRIMARY KEY (queue, batch_id)
11+
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

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

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
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+
"encoding/binary"
19+
"fmt"
20+
"math"
21+
22+
"github.com/uber/submitqueue/submitqueue/extension/conflict/tango"
23+
)
24+
25+
// idBytes is the width of an encoded target ID.
26+
const idBytes = 8
27+
28+
// encodeSignatureTargets packs targets as a flat byte string: per target, the ID
29+
// as 8 big-endian bytes followed by the distance as a signed varint. The result
30+
// is never nil, because a nil slice would be written as NULL into a NOT NULL
31+
// column.
32+
func encodeSignatureTargets(targets []tango.SignatureTarget) []byte {
33+
encoded := make([]byte, 0, len(targets)*(idBytes+1))
34+
for _, target := range targets {
35+
encoded = binary.BigEndian.AppendUint64(encoded, target.ID)
36+
encoded = binary.AppendVarint(encoded, int64(target.Distance))
37+
}
38+
return encoded
39+
}
40+
41+
// decodeSignatureTargets reverses encodeSignatureTargets. It returns an error on
42+
// truncated input or a distance outside the int32 range.
43+
func decodeSignatureTargets(encoded []byte) ([]tango.SignatureTarget, error) {
44+
var targets []tango.SignatureTarget
45+
for offset := 0; offset < len(encoded); {
46+
if len(encoded)-offset < idBytes {
47+
return nil, fmt.Errorf("truncated target ID at offset %d of %d bytes", offset, len(encoded))
48+
}
49+
id := binary.BigEndian.Uint64(encoded[offset:])
50+
offset += idBytes
51+
52+
distance, read := binary.Varint(encoded[offset:])
53+
if read <= 0 {
54+
return nil, fmt.Errorf("invalid distance varint at offset %d of %d bytes", offset, len(encoded))
55+
}
56+
offset += read
57+
if distance < math.MinInt32 || distance > math.MaxInt32 {
58+
return nil, fmt.Errorf("distance %d for target ID %d is outside the int32 range", distance, id)
59+
}
60+
61+
targets = append(targets, tango.SignatureTarget{ID: id, Distance: int32(distance)})
62+
}
63+
return targets, nil
64+
}

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

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
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+
"encoding/binary"
19+
"math"
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+
)
27+
28+
func TestSignatureTargetsCodec_RoundTrip(t *testing.T) {
29+
tests := []struct {
30+
name string
31+
targets []tango.SignatureTarget
32+
}{
33+
{
34+
name: "empty",
35+
targets: nil,
36+
},
37+
{
38+
name: "single target",
39+
targets: []tango.SignatureTarget{{ID: 1, Distance: 0}},
40+
},
41+
{
42+
name: "several targets in ascending ID order",
43+
targets: []tango.SignatureTarget{
44+
{ID: 0x735cc8dafbb93f02, Distance: 0},
45+
{ID: 0x90afc9166e08f639, Distance: 1},
46+
{ID: 0xfcf7cdab318e139e, Distance: 2},
47+
},
48+
},
49+
{
50+
name: "extreme IDs and distances",
51+
targets: []tango.SignatureTarget{
52+
{ID: 0, Distance: math.MaxInt32},
53+
{ID: math.MaxUint64, Distance: 63},
54+
{ID: math.MaxUint64 - 1, Distance: 64},
55+
},
56+
},
57+
{
58+
name: "negative distance",
59+
targets: []tango.SignatureTarget{{ID: 7, Distance: -1}},
60+
},
61+
}
62+
for _, tt := range tests {
63+
t.Run(tt.name, func(t *testing.T) {
64+
got, err := decodeSignatureTargets(encodeSignatureTargets(tt.targets))
65+
require.NoError(t, err)
66+
assert.Equal(t, tt.targets, got)
67+
})
68+
}
69+
}
70+
71+
func TestEncodeSignatureTargets_EmptyIsNotNil(t *testing.T) {
72+
assert.NotNil(t, encodeSignatureTargets(nil))
73+
assert.Empty(t, encodeSignatureTargets(nil))
74+
}
75+
76+
func TestEncodeSignatureTargets_Layout(t *testing.T) {
77+
got := encodeSignatureTargets([]tango.SignatureTarget{{ID: 0x0102030405060708, Distance: 2}})
78+
79+
assert.Equal(t, []byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x04}, got)
80+
}
81+
82+
func TestDecodeSignatureTargets_Invalid(t *testing.T) {
83+
idOnly := binary.BigEndian.AppendUint64(nil, 1)
84+
85+
tests := []struct {
86+
name string
87+
encoded []byte
88+
}{
89+
{
90+
name: "truncated ID",
91+
encoded: []byte{0x01, 0x02, 0x03},
92+
},
93+
{
94+
name: "ID without a distance",
95+
encoded: idOnly,
96+
},
97+
{
98+
name: "unterminated distance varint",
99+
encoded: append(append([]byte{}, idOnly...), 0x80),
100+
},
101+
{
102+
name: "distance outside int32",
103+
encoded: binary.AppendVarint(append([]byte{}, idOnly...), math.MaxInt32+1),
104+
},
105+
{
106+
name: "valid target followed by a truncated one",
107+
encoded: append(binary.AppendVarint(append([]byte{}, idOnly...), 0), 0x01, 0x02),
108+
},
109+
}
110+
for _, tt := range tests {
111+
t.Run(tt.name, func(t *testing.T) {
112+
_, err := decodeSignatureTargets(tt.encoded)
113+
assert.Error(t, err)
114+
})
115+
}
116+
}

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

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,91 @@
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 signatureStore 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+
// NewSignatureStore creates a new MySQL-backed TargetSignatureStore.
40+
func NewSignatureStore(db *sql.DB, scope tally.Scope, queue string) tango.SignatureStore {
41+
return &signatureStore{db: db, scope: scope, queue: queue}
42+
}
43+
44+
// Create inserts a signature. Returns ErrAlreadyExists if the batch already has one.
45+
func (s *signatureStore) Create(ctx context.Context, signature tango.TargetSignature) (retErr error) {
46+
op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets)
47+
defer func() { op.Complete(retErr) }()
48+
49+
if signature.Queue != s.queue {
50+
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)
51+
}
52+
53+
_, err := s.db.ExecContext(ctx,
54+
"INSERT INTO target_signature (queue, batch_id, base_revision, targets, created_at) VALUES (?, ?, ?, ?, ?)",
55+
signature.Queue, signature.BatchID, signature.BaseRevision, encodeSignatureTargets(signature.Targets), signature.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 signature batch_id=%s: %w", signature.BatchID, storage.ErrAlreadyExists)
61+
}
62+
return fmt.Errorf("failed to insert target signature batch_id=%s: %w", signature.BatchID, err)
63+
}
64+
return nil
65+
}
66+
67+
// Get returns the bound queue's signature for batchID. The bool is false when
68+
// the batch has none.
69+
func (s *signatureStore) Get(ctx context.Context, batchID string) (ret tango.TargetSignature, found bool, retErr error) {
70+
op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
71+
defer func() { op.Complete(retErr) }()
72+
73+
signature := tango.TargetSignature{Queue: s.queue, BatchID: batchID}
74+
var encodedTargets []byte
75+
err := s.db.QueryRowContext(ctx,
76+
"SELECT base_revision, targets, created_at FROM target_signature WHERE queue = ? AND batch_id = ?",
77+
s.queue, batchID,
78+
).Scan(&signature.BaseRevision, &encodedTargets, &signature.CreatedAt)
79+
if errors.Is(err, sql.ErrNoRows) {
80+
return tango.TargetSignature{}, false, nil
81+
}
82+
if err != nil {
83+
return tango.TargetSignature{}, false, fmt.Errorf("failed to get target signature batch_id=%s from the database: %w", batchID, err)
84+
}
85+
86+
signature.Targets, err = decodeSignatureTargets(encodedTargets)
87+
if err != nil {
88+
return tango.TargetSignature{}, false, fmt.Errorf("failed to decode targets of target signature batch_id=%s: %w", batchID, err)
89+
}
90+
return signature, true, nil
91+
}

0 commit comments

Comments
 (0)