add change store extension for per-URI request claims

Introduce a dedicated extension that tracks (URI, request_id) claims with mutable
metadata, scoped to queue. Foundation for cross-request URI overlap detection;
no consumers in this commit.

- entity/change_record.go: immutable identity (URI, RequestID), mutable metadata
- extension/changestore/: ChangeStore interface, mysql impl, mock, schema
- INSERT IGNORE on writes for retry idempotency on (uri, request_id)
- FindOverlapping is single-table; callers do their own liveness check
- integration test under test/integration/extension/changestore/mysql/
- e2e suite applies the change schema (harmless no-op until consumers land)

Author: behinddwalls

Makefile

@@ -245,7 +245,7 @@ local-stop: ## Stop all services (keep data)
 
 mocks: ## Generate mock files using mockgen
 	@echo "Generating mocks..."
-	@$(BAZEL) run @rules_go//go -- generate ./extension/storage/... ./extension/counter/... ./extension/queue/... ./extension/mergechecker/... ./extension/scorer/... ./core/consumer/...
+	@$(BAZEL) run @rules_go//go -- generate ./extension/storage/... ./extension/changestore/... ./extension/counter/... ./extension/queue/... ./extension/mergechecker/... ./extension/scorer/... ./core/consumer/...
 	@echo "Mocks generated successfully!"
 
 proto: ## Generate protobuf files from .proto definitions

entity/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "batch_dependent.go",
         "build.go",
         "change_provider.go",
+        "change_record.go",
         "land_request.go",
         "queue_config.go",
         "request.go",

entity/change_record.go

@@ -0,0 +1,45 @@
+// 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
+
+// ChangeRecord represents a single URI's claim by a request, persisted in the change store.
+// The (URI, RequestID) pair is the identity and is immutable; Metadata may be updated over
+// time as additional information about the change (e.g., PR title, author, mergeability)
+// becomes available.
+type ChangeRecord struct {
+	// URI identifies the change (RFC 3986). Same scheme/format as entity.Change.URIs.
+	// Example: "github://uber/submitqueue/pull/123/abc123def".
+	URI string `json:"uri"`
+
+	// RequestID is the owning land request that claimed this URI.
+	// Format matches entity.Request.ID: "<queue>/<counter_value>".
+	RequestID string `json:"request_id"`
+
+	// Queue is the queue scope for the owning request. Denormalized from the request
+	// to allow queue-scoped duplicate checks without a join.
+	Queue string `json:"queue"`
+
+	// Metadata is a JSON-encoded blob of provider-specific information about the change
+	// (e.g., PR title, author, mergeable state). Empty when the record is first written;
+	// populated and updated by downstream enrichment.
+	Metadata string `json:"metadata,omitempty"`
+
+	// CreatedAt is the Unix milliseconds timestamp when this record was first created.
+	CreatedAt int64 `json:"created_at"`
+
+	// UpdatedAt is the Unix milliseconds timestamp when this record's Metadata was last updated.
+	// Equal to CreatedAt when the record has never been updated.
+	UpdatedAt int64 `json:"updated_at"`
+}

extension/changestore/BUILD.bazel

@@ -0,0 +1,9 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "changestore",
+    srcs = ["change_store.go"],
+    importpath = "github.com/uber/submitqueue/extension/changestore",
+    visibility = ["//visibility:public"],
+    deps = ["//entity"],
+)

extension/changestore/README.md

@@ -0,0 +1,19 @@
+# ChangeStore
+
+Vendor-agnostic interface for tracking per-URI claims by in-flight land requests.
+
+Each record asserts that a specific URI (e.g., a GitHub PR) was claimed by a specific request, scoped to a queue. The store is consulted by the orchestrator's `start` controller to detect duplicate requests — submissions whose URIs overlap with another in-flight request's URIs in the same queue.
+
+## Semantics
+
+- **Identity is immutable.** A record is keyed by `(URI, RequestID)`; once written, that pair is never mutated.
+- **Metadata is mutable.** The `Metadata` field is intended for provider-specific information about the change (PR title, author, mergeability, etc.) that may be enriched after the record is first written. `UpdatedAt` reflects the last metadata change; `CreatedAt` is fixed at write time.
+- **Idempotent writes.** `Create` ignores primary-key conflicts so queue-redelivery of the same request is a safe no-op.
+- **No liveness filter.** `FindOverlapping` returns records regardless of whether the owning request is still in flight. Callers must check liveness against `RequestStore` themselves — the store boundary is intentionally one query, one table, no joins.
+- **Append-only by design.** Records are not deleted when the owning request reaches a terminal state; the historical claim is preserved for audit. Duplicate detection filters terminals out at query time via the controller-side liveness check.
+
+## Implementing a Backend
+
+1. Create `extension/changestore/{backend}/` directory.
+2. Implement the `ChangeStore` interface.
+3. Add schema files under `extension/changestore/{backend}/schema/` if the backend requires them.

extension/changestore/change_store.go

@@ -0,0 +1,51 @@
+// 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 changestore
+
+//go:generate mockgen -source=change_store.go -destination=mock/change_store_mock.go -package=mock
+
+import (
+	"context"
+
+	"github.com/uber/submitqueue/entity"
+)
+
+// ChangeStore manages per-URI claim records for in-flight land requests.
+// Each row records that a specific URI was claimed by a specific request, scoped to a queue.
+// The (URI, RequestID) pair is the immutable identity of a record. Metadata may evolve over time.
+//
+// The store is the source of truth for "which URIs are or were associated with which requests".
+// Liveness of an owning request is NOT tracked here — callers must consult RequestStore separately
+// to determine whether an owner is in a terminal state.
+type ChangeStore interface {
+	// Create persists a batch of ChangeRecords. Conflicts on the (URI, RequestID) primary key
+	// are silently ignored, making the call idempotent under queue redeliveries of the same request.
+	// Records belonging to different requests do not conflict — overlap between requests is detected
+	// by FindOverlapping, not by Create.
+	Create(ctx context.Context, records []entity.ChangeRecord) error
+
+	// FindOverlapping returns ChangeRecords whose URI is in the given set, scoped to queue,
+	// excluding any records belonging to excludeRequestID (so callers can skip self when checking
+	// for duplicates of a freshly-claimed request). Returns an empty slice when there is no overlap.
+	//
+	// Liveness of the returned records' owning requests is NOT filtered here — the caller is
+	// responsible for consulting RequestStore to skip terminal owners.
+	FindOverlapping(
+		ctx context.Context,
+		queue string,
+		uris []string,
+		excludeRequestID string,
+	) ([]entity.ChangeRecord, error)
+}

extension/changestore/mock/BUILD.bazel

@@ -0,0 +1,12 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "mock",
+    srcs = ["change_store_mock.go"],
+    importpath = "github.com/uber/submitqueue/extension/changestore/mock",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//entity",
+        "@org_uber_go_mock//gomock",
+    ],
+)

extension/changestore/mock/change_store_mock.go

@@ -0,0 +1,14 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "mysql",
+    srcs = ["change_store.go"],
+    importpath = "github.com/uber/submitqueue/extension/changestore/mysql",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//core/metrics",
+        "//entity",
+        "//extension/changestore",
+        "@com_github_uber_go_tally_v4//:tally",
+    ],
+)

extension/changestore/mysql/BUILD.bazel

@@ -0,0 +1,118 @@
+// 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"
+	"fmt"
+	"strings"
+
+	"github.com/uber-go/tally/v4"
+
+	"github.com/uber/submitqueue/core/metrics"
+	"github.com/uber/submitqueue/entity"
+	"github.com/uber/submitqueue/extension/changestore"
+)
+
+type changeStore struct {
+	db    *sql.DB
+	scope tally.Scope
+}
+
+// NewChangeStore creates a new MySQL-backed ChangeStore.
+func NewChangeStore(db *sql.DB, scope tally.Scope) changestore.ChangeStore {
+	return &changeStore{db: db, scope: scope}
+}
+
+// Create inserts a batch of ChangeRecords. Primary-key conflicts on (uri, request_id)
+// are silently ignored via INSERT IGNORE so queue-redelivery of the same request is a no-op.
+func (s *changeStore) Create(ctx context.Context, records []entity.ChangeRecord) (retErr error) {
+	op := metrics.Begin(s.scope, "create")
+	defer func() { op.Complete(retErr) }()
+
+	if len(records) == 0 {
+		return nil
+	}
+
+	const cols = 6
+	placeholders := strings.Repeat("(?, ?, ?, ?, ?, ?), ", len(records))
+	placeholders = placeholders[:len(placeholders)-2] // trim trailing ", "
+
+	args := make([]any, 0, len(records)*cols)
+	for _, r := range records {
+		// Pass empty Metadata as NULL — JSON column rejects empty string but accepts NULL.
+		var metadata any
+		if r.Metadata != "" {
+			metadata = r.Metadata
+		}
+		args = append(args, r.URI, r.RequestID, r.Queue, metadata, r.CreatedAt, r.UpdatedAt)
+	}
+
+	query := "INSERT IGNORE INTO `change` (uri, request_id, queue, metadata, created_at, updated_at) VALUES " + placeholders
+	if _, err := s.db.ExecContext(ctx, query, args...); err != nil {
+		return fmt.Errorf("failed to insert change records (count=%d): %w", len(records), err)
+	}
+	return nil
+}
+
+// FindOverlapping returns ChangeRecords whose uri is in the given set, scoped to queue,
+// excluding any belonging to excludeRequestID.
+func (s *changeStore) FindOverlapping(
+	ctx context.Context,
+	queue string,
+	uris []string,
+	excludeRequestID string,
+) (ret []entity.ChangeRecord, retErr error) {
+	op := metrics.Begin(s.scope, "find_overlapping")
+	defer func() { op.Complete(retErr) }()
+
+	if len(uris) == 0 {
+		return nil, nil
+	}
+
+	uriPlaceholders := "?" + strings.Repeat(", ?", len(uris)-1)
+	query := "SELECT uri, request_id, queue, metadata, created_at, updated_at FROM `change` " +
+		"WHERE uri IN (" + uriPlaceholders + ") AND queue = ? AND request_id != ?"
+
+	args := make([]any, 0, len(uris)+2)
+	for _, u := range uris {
+		args = append(args, u)
+	}
+	args = append(args, queue, excludeRequestID)
+
+	rows, err := s.db.QueryContext(ctx, query, args...)
+	if err != nil {
+		return nil, fmt.Errorf("failed to query overlapping changes for queue=%s: %w", queue, err)
+	}
+	defer rows.Close()
+
+	var results []entity.ChangeRecord
+	for rows.Next() {
+		var rec entity.ChangeRecord
+		var metadata sql.NullString
+		if err := rows.Scan(&rec.URI, &rec.RequestID, &rec.Queue, &metadata, &rec.CreatedAt, &rec.UpdatedAt); err != nil {
+			return nil, fmt.Errorf("failed to scan change record for queue=%s: %w", queue, err)
+		}
+		if metadata.Valid {
+			rec.Metadata = metadata.String
+		}
+		results = append(results, rec)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("failed to iterate change records for queue=%s: %w", queue, err)
+	}
+	return results, nil
+}