Skip to content
Open
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
41 changes: 41 additions & 0 deletions submitqueue/extension/conflict/tango/mysql/BUILD.bazel
Original file line number Diff line number Diff line change
@@ -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",
],
)
19 changes: 19 additions & 0 deletions submitqueue/extension/conflict/tango/mysql/mysql.go
Original file line number Diff line number Diff line change
@@ -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
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
filegroup(
name = "schema",
srcs = glob(["*.sql"]),
visibility = ["//visibility:public"],
)
Original file line number Diff line number Diff line change
@@ -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;
64 changes: 64 additions & 0 deletions submitqueue/extension/conflict/tango/mysql/signature_codec.go
Original file line number Diff line number Diff line change
@@ -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
}
116 changes: 116 additions & 0 deletions submitqueue/extension/conflict/tango/mysql/signature_codec_test.go
Original file line number Diff line number Diff line change
@@ -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)
})
}
}
91 changes: 91 additions & 0 deletions submitqueue/extension/conflict/tango/mysql/signature_store.go
Original file line number Diff line number Diff line change
@@ -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
}
Loading
Loading