fix(orchestrator): reconcile batch redelivery

submitqueue/entity/BUILD.bazel

@@ -18,6 +18,7 @@ go_library(
         "queue.go",
         "queue_config.go",
         "request.go",
+        "request_batch.go",
         "request_history.go",
         "request_log.go",
         "request_summary.go",

submitqueue/entity/request_batch.go

@@ -0,0 +1,27 @@
+// 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 entity
+
+// RequestBatch is the immutable assignment of one request to one batch.
+// RequestID is unique, so retries resolve the same BatchID instead of creating
+// another logical batch for the request.
+type RequestBatch struct {
+	// RequestID is the globally unique request identifier.
+	RequestID string
+	// BatchID is the globally unique batch identifier assigned to the request.
+	BatchID string
+	// Version is the version of the assignment. Immutable assignments start at 1.
+	Version int32
+}

submitqueue/extension/storage/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "batch_store.go",
         "build_store.go",
         "change_store.go",
+        "request_batch_store.go",
         "request_log_store.go",
         "request_queue_summary_store.go",
         "request_store.go",

submitqueue/extension/storage/mock/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "batch_store_mock.go",
         "build_store_mock.go",
         "change_store_mock.go",
+        "request_batch_store_mock.go",
         "request_log_store_mock.go",
         "request_queue_summary_store_mock.go",
         "request_store_mock.go",

submitqueue/extension/storage/mysql/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "batch_store.go",
         "build_store.go",
         "change_store.go",
+        "request_batch_store.go",
         "request_log_store.go",
         "request_queue_summary_store.go",
         "request_store.go",
@@ -34,6 +35,7 @@ go_test(
         "batch_store_test.go",
         "build_store_test.go",
         "change_store_test.go",
+        "request_batch_store_test.go",
         "request_log_store_test.go",
         "request_queue_summary_store_test.go",
         "request_store_test.go",

submitqueue/extension/storage/mysql/request_batch_store.go

@@ -0,0 +1,76 @@
+// 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/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 requestBatchStore struct {
+	db    *sql.DB
+	scope tally.Scope
+}
+
+// NewRequestBatchStore creates a new MySQL-backed RequestBatchStore.
+func NewRequestBatchStore(db *sql.DB, scope tally.Scope) storage.RequestBatchStore {
+	return &requestBatchStore{db: db, scope: scope}
+}
+
+// Get retrieves the immutable batch assignment for a request.
+func (s *requestBatchStore) Get(ctx context.Context, requestID string) (ret entity.RequestBatch, retErr error) {
+	op := metrics.Begin(s.scope, "get")
+	defer func() { op.Complete(retErr) }()
+
+	err := s.db.QueryRowContext(ctx,
+		"SELECT request_id, batch_id, version FROM request_batch WHERE request_id = ?",
+		requestID,
+	).Scan(&ret.RequestID, &ret.BatchID, &ret.Version)
+	if errors.Is(err, sql.ErrNoRows) {
+		return entity.RequestBatch{}, storage.WrapNotFound(err)
+	}
+	if err != nil {
+		return entity.RequestBatch{}, fmt.Errorf("failed to get request batch assignment requestID=%s: %w", requestID, err)
+	}
+	return ret, nil
+}
+
+// Create reserves an immutable batch assignment for a request.
+func (s *requestBatchStore) Create(ctx context.Context, requestBatch entity.RequestBatch) (retErr error) {
+	op := metrics.Begin(s.scope, "create")
+	defer func() { op.Complete(retErr) }()
+
+	_, err := s.db.ExecContext(ctx,
+		"INSERT INTO request_batch (request_id, batch_id, version) VALUES (?, ?, ?)",
+		requestBatch.RequestID, requestBatch.BatchID, requestBatch.Version,
+	)
+	if err != nil {
+		var mysqlErr *mysql.MySQLError
+		if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry {
+			return fmt.Errorf("request batch assignment requestID=%s: %w", requestBatch.RequestID, storage.ErrAlreadyExists)
+		}
+		return fmt.Errorf("failed to create request batch assignment requestID=%s batchID=%s: %w", requestBatch.RequestID, requestBatch.BatchID, err)
+	}
+	return nil
+}