feat(storage): Add MySQL backed target signature store

Author: manjari25

doc/rfc/service-scoped-extensions.md

@@ -25,7 +25,7 @@ Separating the services onto their own databases already worked — the orchestr
 1. An extension whose **aggregate** is used by exactly one service of a multi-service domain puts that aggregate, its implementations, its mocks and its schema at `{domain}/{service}/extension/{ext}/`. The store contracts themselves stay at `{domain}/extension/{ext}/`: the aggregate is what decides reachability, so moving the contracts adds no enforcement while forcing domain-level callers to import a service package.
 2. An extension shared between a domain's services stays wholly at `{domain}/extension/{ext}/` — including one used by a single service today but expected in both. `submitqueue/extension/queueconfig/` is gateway-only now and stays where it is on that basis. A single-service domain is unaffected: its domain root is its service root, so nothing moves.
 3. Storage is the extension this split applied to. `changeprovider`, `validator`, `conflict`, `buildrunner` and `speculation` are orchestrator-only and would qualify under Decision 1, but none has a schema and none was the reason for this change; they stay at `{domain}/extension/` until a later RFC takes them. The trigger for moving storage was a schema that a separate-database deployment must not over-provision.
-4. SubmitQueue's storage aggregate splits in two: `submitqueue/gateway/extension/storage/` declares a `Factory` and a four-accessor `Storage`, `submitqueue/orchestrator/extension/storage/` a `Factory` and a nine-accessor `Storage`. Neither can resolve the other's stores.
+4. SubmitQueue's storage aggregate splits in two: `submitqueue/gateway/extension/storage/` declares a `Factory` and a four-accessor `Storage`, `submitqueue/orchestrator/extension/storage/` a `Factory` and a ten-accessor `Storage`. Neither can resolve the other's stores.
 5. `submitqueue/extension/storage/` keeps the fifteen store contracts, their mocks, the error vocabulary, `Config`, and the package README, which documents the contracts. Its import path does not change, so callers that name only a contract or an error are untouched.
 6. The contract package is not promoted to `platform/`. It is not cross-domain — `stovepipe/extension/storage/storage.go` already declares its own verbatim copy of the same symbols, so a per-domain error vocabulary is the existing pattern rather than something this RFC introduces.
 7. Each half owns its own MySQL schema directory, so the tables a service creates are the tables it reads. A union target over both stays available for deployments that colocate the services — the end-to-end tests and the local compose stack take it and name no service directory. Split schemas are the opt-in, not the default.
@@ -52,6 +52,7 @@ Every store, verified against actual usage rather than intent.
 | `BuildStore` | `build` |
 | `SpeculationPathSetStore` | `speculation_path_set` |
 | `PathBuildStore` | `path_build` |
+| `TargetSignatureStore` | `target_signature` |
 
 `counter` is not in either list. It is its own shared extension and each service keeps its own counter table.
 

submitqueue/orchestrator/extension/storage/mock/storage_mock.go

@@ -13,6 +13,8 @@ go_library(
         "request_store.go",
         "speculation_path_set_store.go",
         "storage.go",
+        "target_signature_codec.go",
+        "target_signature_store.go",
     ],
     importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mysql",
     visibility = ["//visibility:public"],
@@ -38,6 +40,8 @@ go_test(
         "request_store_test.go",
         "speculation_path_set_store_test.go",
         "storage_test.go",
+        "target_signature_codec_test.go",
+        "target_signature_store_test.go",
     ],
     embed = [":go_default_library"],
     deps = [

submitqueue/orchestrator/extension/storage/mysql/BUILD.bazel

@@ -2,7 +2,7 @@
 
 ## Queue-leading primary keys
 
-Every table leads its primary key with `queue`: `request` and `batch` on `(queue, id)`, `build` on `(queue, id)`, `batch_dependent` on `(queue, batch_id)`, `request_batch` on `(queue, request_id, batch_id)`, `change` on `(queue, uri, request_id)`, `queue_batch_state` on `(queue, state, batch_id)`, and `speculation_path_set` on `(queue, head)`. A queue-bound store instance prefixes every read and stamps every write with its bound queue, so one queue's rows are unreachable through another queue's binding and every table is shardable by queue. `//tool/linter/queueshard` enforces this, and also rejects any secondary index that does not itself lead with `queue`, since such an index would reintroduce a cross-queue access path.
+Every table leads its primary key with `queue`: `request` and `batch` on `(queue, id)`, `build` on `(queue, id)`, `batch_dependent` on `(queue, batch_id)`, `request_batch` on `(queue, request_id, batch_id)`, `change` on `(queue, uri, request_id)`, `queue_batch_state` on `(queue, state, batch_id)`, `speculation_path_set` on `(queue, head)`, and `target_signature` on `(queue, batch_id)`. A queue-bound store instance prefixes every read and stamps every write with its bound queue, so one queue's rows are unreachable through another queue's binding and every table is shardable by queue. `//tool/linter/queueshard` enforces this, and also rejects any secondary index that does not itself lead with `queue`, since such an index would reintroduce a cross-queue access path.
 
 The `build` key also removes a cross-queue uniqueness assumption: build IDs are runner-minted, so two queues sharing one CI pipeline may legitimately mint the same identifier. `speculation_path_set` relies on the same property for its head: a batch ID is unique only within its queue.
 
@@ -27,3 +27,7 @@ Terminal-state records (and their batches) accumulate as the queue processes wor
 ### Composite primary key: `(queue, uri, request_id)`
 
 The `change` table records per-URI claims by in-flight requests. `request_id` is part of the primary key so that concurrent claims on the same URI by different requests coexist as distinct rows — a same-request retry collides on the PK and is a no-op (`INSERT IGNORE`), while a different-request claim is a new row that `GetByURI` surfaces for overlap detection. `queue` leads the key so queue-scoped lookups are primary-key-prefix scans and the table is shardable by queue.
+
+## target_signature table
+
+`target_signature` holds one row per batch: the base revision the signature was computed at and the batch's affected targets as sorted `(id, distance)` entries (see `entity.TargetSignature`). Rows are written once and never updated. `targets` is a flat binary encoding, per entry an 8-byte big-endian ID followed by a signed varint distance, stored as `LONGBLOB` because the largest signatures run to tens of megabytes. The encoding lives in `target_signature_codec.go`. MySQL's `max_allowed_packet` is a separate limit and must be sized for the largest signature a deployment expects.

submitqueue/orchestrator/extension/storage/mysql/schema/README.md

@@ -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;

submitqueue/orchestrator/extension/storage/mysql/schema/target_signature.sql

@@ -60,6 +60,7 @@ func (s *Storage) For(queueName string) (storage.Storage, error) {
 		buildStore:              NewBuildStore(s.db, s.scope.SubScope("build_store"), queueName),
 		speculationPathSetStore: NewSpeculationPathSetStore(s.db, s.scope.SubScope("speculation_path_set_store"), queueName),
 		pathBuildStore:          NewPathBuildStore(s.db, s.scope.SubScope("path_build_store"), queueName),
+		targetSignatureStore:    NewTargetSignatureStore(s.db, s.scope.SubScope("target_signature_store"), queueName),
 	}, nil
 }
 
@@ -79,6 +80,7 @@ type boundStorage struct {
 	buildStore              basestorage.BuildStore
 	speculationPathSetStore basestorage.SpeculationPathSetStore
 	pathBuildStore          basestorage.PathBuildStore
+	targetSignatureStore    basestorage.TargetSignatureStore
 }
 
 // Verify boundStorage implements the queue-scoped aggregate at compile time.
@@ -128,3 +130,8 @@ func (f *boundStorage) GetSpeculationPathSetStore() basestorage.SpeculationPathS
 func (f *boundStorage) GetPathBuildStore() basestorage.PathBuildStore {
 	return f.pathBuildStore
 }
+
+// GetTargetSignatureStore returns the bound MySQL-backed TargetSignatureStore.
+func (f *boundStorage) GetTargetSignatureStore() basestorage.TargetSignatureStore {
+	return f.targetSignatureStore
+}

submitqueue/orchestrator/extension/storage/mysql/storage.go

@@ -46,6 +46,7 @@ func TestNewStorage(t *testing.T) {
 	assert.NotNil(t, bound.GetQueueBatchStateStore())
 	assert.NotNil(t, bound.GetBuildStore())
 	assert.NotNil(t, bound.GetSpeculationPathSetStore())
+	assert.NotNil(t, bound.GetTargetSignatureStore())
 
 	_, err = s.For("")
 	assert.Error(t, err, "resolving an empty queue name must fail")

submitqueue/orchestrator/extension/storage/mysql/storage_test.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/entity"
+)
+
+// 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 []entity.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) ([]entity.SignatureTarget, error) {
+	var targets []entity.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, entity.SignatureTarget{ID: id, Distance: int32(distance)})
+	}
+	return targets, nil
+}

submitqueue/orchestrator/extension/storage/mysql/target_signature_codec.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/entity"
+)
+
+func TestSignatureTargetsCodec_RoundTrip(t *testing.T) {
+	tests := []struct {
+		name    string
+		targets []entity.SignatureTarget
+	}{
+		{
+			name:    "empty",
+			targets: nil,
+		},
+		{
+			name:    "single target",
+			targets: []entity.SignatureTarget{{ID: 1, Distance: 0}},
+		},
+		{
+			name: "several targets in ascending ID order",
+			targets: []entity.SignatureTarget{
+				{ID: 0x735cc8dafbb93f02, Distance: 0},
+				{ID: 0x90afc9166e08f639, Distance: 1},
+				{ID: 0xfcf7cdab318e139e, Distance: 2},
+			},
+		},
+		{
+			name: "extreme IDs and distances",
+			targets: []entity.SignatureTarget{
+				{ID: 0, Distance: math.MaxInt32},
+				{ID: math.MaxUint64, Distance: 63},
+				{ID: math.MaxUint64 - 1, Distance: 64},
+			},
+		},
+		{
+			name:    "negative distance",
+			targets: []entity.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([]entity.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)
+		})
+	}
+}

submitqueue/orchestrator/extension/storage/mysql/target_signature_codec_test.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/entity"
+	"github.com/uber/submitqueue/submitqueue/extension/storage"
+)
+
+type targetSignatureStore 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
+}
+
+// NewTargetSignatureStore creates a new MySQL-backed TargetSignatureStore.
+func NewTargetSignatureStore(db *sql.DB, scope tally.Scope, queue string) storage.TargetSignatureStore {
+	return &targetSignatureStore{db: db, scope: scope, queue: queue}
+}
+
+// Create inserts a signature. Returns ErrAlreadyExists if the batch already has one.
+func (s *targetSignatureStore) Create(ctx context.Context, signature entity.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 *targetSignatureStore) Get(ctx context.Context, batchID string) (ret entity.TargetSignature, found bool, retErr error) {
+	op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
+
+	signature := entity.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 entity.TargetSignature{}, false, nil
+	}
+	if err != nil {
+		return entity.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 entity.TargetSignature{}, false, fmt.Errorf("failed to decode targets of target signature batch_id=%s: %w", batchID, err)
+	}
+	return signature, true, nil
+}