uber · mnoah1 · Jul 7, 2026 · Jul 8, 2026
@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library")
 go_library(
     name = "go_default_library",
     srcs = [
+        "queue_store.go",
         "request_store.go",
         "request_uri_store.go",
         "storage.go",

@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library")
 go_library(
     name = "go_default_library",
     srcs = [
+        "queue_store_mock.go",
         "request_store_mock.go",
         "request_uri_store_mock.go",
         "storage_mock.go",

@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library")
 go_library(
     name = "go_default_library",
     srcs = [
+        "queue_store.go",
         "request_store.go",
         "request_uri_store.go",
         "storage.go",

@@ -0,0 +1,153 @@
+// Copyright (c) 2025 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/uber-go/tally"
+	"github.com/uber/submitqueue/platform/metrics"
+	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/storage"
+)
+
+type queueStore struct {
+	db    *sql.DB
+	scope tally.Scope
+}
+
+// NewQueueStore creates a new MySQL-backed QueueStore.
+func NewQueueStore(db *sql.DB, scope tally.Scope) storage.QueueStore {
+	return &queueStore{db: db, scope: scope}
+}
+
+// GetOrCreate returns the queue row for name, creating it with defaults if absent.
+func (q *queueStore) GetOrCreate(ctx context.Context, name string, defaults entity.Queue) (ret entity.Queue, retErr error) {
+	op := metrics.Begin(q.scope, "get_or_create")
+	defer func() { op.Complete(retErr) }()
+
+	queue, err := q.Get(ctx, name)
+	if err == nil {
+		return queue, nil
+	}
+	if !storage.IsNotFound(err) {
+		return entity.Queue{}, err
+	}
+
+	toCreate := defaults
+	toCreate.Name = name
+	if toCreate.Version == 0 {
+		toCreate.Version = 1
+	}
+
+	if err := q.create(ctx, toCreate); err != nil {
+		if errors.Is(err, storage.ErrAlreadyExists) {
+			return q.Get(ctx, name)
+		}
+		return entity.Queue{}, err
+	}
+	return toCreate, nil
+}
+
+// Get retrieves a queue by name. Returns ErrNotFound if the queue is not found.
+func (q *queueStore) Get(ctx context.Context, name string) (ret entity.Queue, retErr error) {
+	op := metrics.Begin(q.scope, "get")
+	defer func() { op.Complete(retErr) }()
+
+	var queue entity.Queue
+	err := q.db.QueryRowContext(ctx,
+		"SELECT name, last_green_uri, in_flight_count, latest_request_seq, version FROM queue WHERE name = ?",
+		name,
+	).Scan(
+		&queue.Name,
+		&queue.LastGreenURI,
+		&queue.InFlightCount,
+		&queue.LatestRequestSeq,
+		&queue.Version,
+	)
+
+	if errors.Is(err, sql.ErrNoRows) {
+		return entity.Queue{}, storage.WrapNotFound(err)
+	}
+	if err != nil {
+		return entity.Queue{}, fmt.Errorf("failed to get queue name=%s from the database: %w", name, err)
+	}
+
+	return queue, nil
+}
+
+// Update persists the mutable fields of queue if the stored version matches oldVersion,
+// writing newVersion. Returns ErrVersionMismatch if the stored version does not match.
+func (q *queueStore) Update(ctx context.Context, queue entity.Queue, oldVersion, newVersion int32) (retErr error) {
+	op := metrics.Begin(q.scope, "update")
+	defer func() { op.Complete(retErr) }()
+
+	result, err := q.db.ExecContext(ctx,
+		`UPDATE queue
+		 SET last_green_uri = ?, in_flight_count = ?, latest_request_seq = ?, version = ?
+		 WHERE name = ? AND version = ?`,
+		queue.LastGreenURI,
+		queue.InFlightCount,
+		queue.LatestRequestSeq,
+		newVersion,
+		queue.Name,
+		oldVersion,
+	)
+	if err != nil {
+		return fmt.Errorf(
+			"failed to update queue name=%q oldVersion=%d newVersion=%d: %w",
+			queue.Name, oldVersion, newVersion, err,
+		)
+	}
+
+	rowsAffected, err := result.RowsAffected()
+	if err != nil {
+		return fmt.Errorf(
+			"failed to get rows affected from update for name=%q oldVersion=%d newVersion=%d: %w",
+			queue.Name, oldVersion, newVersion, err,
+		)
+	}
+
+	if rowsAffected != 1 {
+		return fmt.Errorf(
+			"version mismatch for queue update: name=%q expected_version=%d: %w",
+			queue.Name, oldVersion, storage.ErrVersionMismatch,
+		)
+	}
+
+	return nil
+}
+
+func (q *queueStore) create(ctx context.Context, queue entity.Queue) error {
+	_, err := q.db.ExecContext(ctx,
+		`INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_seq, version)
+		 VALUES (?, ?, ?, ?, ?)`,
+		queue.Name,
+		queue.LastGreenURI,
+		queue.InFlightCount,
+		queue.LatestRequestSeq,
+		queue.Version,
+	)
+	if err != nil {
+		if isDuplicateEntry(err) {
+			return fmt.Errorf("queue name=%s: %w", queue.Name, storage.ErrAlreadyExists)
+		}
+		return fmt.Errorf("failed to insert queue name=%s: %w", queue.Name, err)
+	}
+	return nil
+}
@@ -0,0 +1,10 @@
+-- queue holds per-queue coordination state for the validation pipeline: the last-green
+-- bookmark, in-flight gate count, and latest-request sequence pointer.
+CREATE TABLE IF NOT EXISTS queue (
+    name               VARCHAR(255) NOT NULL,
+    last_green_uri     VARCHAR(255) NOT NULL DEFAULT '',
+    in_flight_count    INT          NOT NULL DEFAULT 0,
+    latest_request_seq BIGINT       NOT NULL DEFAULT 0,
+    version            INT          NOT NULL,
+    PRIMARY KEY (name)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
@@ -19,14 +19,14 @@ import (
 
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/uber-go/tally"
-
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 )
 
 type mysqlStorage struct {
 	db              *sql.DB
 	requestStore    storage.RequestStore
 	requestURIStore storage.RequestURIStore
+	queueStore      storage.QueueStore
 }
 
 // NewStorage creates a new MySQL-backed storage.
@@ -35,6 +35,7 @@ func NewStorage(db *sql.DB, scope tally.Scope) (storage.Storage, error) {
 		db:              db,
 		requestStore:    NewRequestStore(db, scope.SubScope("request_store")),
 		requestURIStore: NewRequestURIStore(db, scope.SubScope("request_uri_store")),
+		queueStore:      NewQueueStore(db, scope.SubScope("queue_store")),
 	}, nil
 }
 
@@ -48,6 +49,11 @@ func (f *mysqlStorage) GetRequestURIStore() storage.RequestURIStore {
 	return f.requestURIStore
 }
 
+// GetQueueStore returns the MySQL-backed QueueStore.
+func (f *mysqlStorage) GetQueueStore() storage.QueueStore {
+	return f.queueStore
+}
+
 // Close closes the underlying database connection.
 func (f *mysqlStorage) Close() error {
 	return f.db.Close()

@@ -0,0 +1,43 @@
+// Copyright (c) 2025 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 storage
+
+//go:generate mockgen -source=queue_store.go -destination=mock/queue_store_mock.go -package=mock
+
+import (
+	"context"
+
+	"github.com/uber/submitqueue/stovepipe/entity"
+)
+
+// QueueStore persists per-queue coordination rows, keyed by queue name.
+type QueueStore interface {
+	// GetOrCreate returns the queue row for name, creating it with defaults if absent.
+	// defaults supplies initial field values for a new row (Name is set from name).
+	// On a create race, re-reads and returns the canonical row.
+	GetOrCreate(ctx context.Context, name string, defaults entity.Queue) (entity.Queue, error)
+
+	// Get retrieves a queue by name. Returns ErrNotFound if the queue is not found.
+	Get(ctx context.Context, name string) (entity.Queue, error)
+
+	// Update persists the mutable fields of queue if the stored version matches
+	// oldVersion, writing newVersion. Returns ErrVersionMismatch if the stored
+	// version does not match (including when the queue does not exist).
+	//
+	// Version arithmetic is owned by the caller: it computes newVersion (typically
+	// oldVersion+1) and only assigns queue.Version = newVersion after this call
+	// succeeds. The store performs a pure conditional write.
+	Update(ctx context.Context, queue entity.Queue, oldVersion, newVersion int32) error
+}
@@ -50,6 +50,9 @@ type Storage interface {
 	// GetRequestURIStore returns the RequestURIStore instance.
 	GetRequestURIStore() RequestURIStore
 
+	// GetQueueStore returns the QueueStore instance.
+	GetQueueStore() QueueStore
+
 	// Close closes the storage and all underlying connections. Should only be called once at the end of the program.
 	Close() error
 }