address review comments: queue-leading PK, NOT NULL metadata, cleaner API

- schema: PRIMARY KEY (queue, uri, request_id) — queue-scoped lookups become
  PK-prefix scans and the table is shardable by queue. Comment in the schema
  explains why request_id stays in the PK (concurrent claims by different
  requests coexist as distinct rows; same-request retries collide on the PK).
- schema: metadata JSON NOT NULL. The mysql impl writes '{}' for empty
  metadata so callers don't need to know about the column constraint.
- interface: drop excludeRequestID from FindOverlapping. Callers that want to
  skip self filter the result by RequestID themselves. Documented Create's
  batch atomicity (single multi-row INSERT, all-or-nothing).
- mysql: FindOverlapping query now leads with `WHERE queue = ?` to align with
  the new PK order.
- entity: expanded RequestID/Queue field comments to explain their PK roles.
- README: documents the new key shape and metadata semantics.
- integration tests: drop the 4th arg, replace TestFindOverlapping_ExcludesSelf
  with a test that asserts the store does NOT exclude self, and add an
  assertion that empty metadata round-trips as '{}'.

Author: behinddwalls

entity/change_record.go

@@ -15,25 +15,31 @@
 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.
+// The (Queue, URI, RequestID) triple 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 participates in the change-store primary key so that concurrent claims
+	// by different requests on the same URI coexist as distinct rows. Same-request
+	// retries collide on the PK and are absorbed idempotently; different-request
+	// collisions surface as additional rows that callers detect via FindOverlapping.
 	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 is the queue the owning request belongs to. It is the leading column of
+	// the change-store primary key, so queue-scoped duplicate checks become PK-prefix
+	// scans and the table is shardable by queue.
 	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.
+	// (e.g., PR title, author, mergeable state). Stored as `'{}'` when no metadata has
+	// been populated yet; updated by downstream enrichment.
 	Metadata string `json:"metadata,omitempty"`
 
 	// CreatedAt is the Unix milliseconds timestamp when this record was first created.

extension/changestore/README.md

@@ -6,10 +6,12 @@ Each record asserts that a specific URI (e.g., a GitHub PR) was claimed by a spe
 
 ## 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.
+- **Identity is immutable.** A record is keyed by `(Queue, URI, RequestID)`; once written, that triple is never mutated.
+- **Queue leads the key.** Backends should make `Queue` the leading column of the primary key (or partition key, in shardable stores). All reads are queue-scoped, so this turns lookups into PK-prefix scans and keeps the table shardable.
+- **`RequestID` in the key is intentional.** Concurrent claims by different requests on the same URI coexist as distinct rows. Same-request retries collide on the PK and are absorbed idempotently; cross-request collisions show up as additional rows that callers detect via `FindOverlapping`.
+- **Metadata is required and mutable.** The `Metadata` field is JSON. The store treats `'{}'` as the canonical "no metadata yet" value — callers that pass an empty Go string get `'{}'` written. Downstream enrichment can update it; `UpdatedAt` reflects the last update.
+- **Idempotent writes, atomic batches.** `Create` ignores primary-key conflicts so queue-redelivery of the same request is a safe no-op. The whole batch is one underlying multi-row INSERT — partial success is not exposed.
+- **No filtering at the store layer.** `FindOverlapping` returns every matching row, including ones owned by the caller's own request. Callers that want to skip self filter the result by `RequestID` themselves. Liveness is also the caller's job — consult `RequestStore` to skip terminal owners. 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

extension/changestore/change_store.go

@@ -30,22 +30,29 @@ import (
 // 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
+	// Create persists a batch of ChangeRecords as a single atomic operation: either all
+	// records are written or none are. The batch corresponds to one underlying multi-row
+	// INSERT, so partial success is never observable.
+	//
+	// Primary-key conflicts on (queue, uri, request_id) are silently ignored, which makes
+	// the call idempotent under queue redeliveries of the same request. Records belonging
+	// to different requests do not conflict on the PK — cross-request overlap 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.
+	// FindOverlapping returns ChangeRecords whose URI is in the given set, scoped to queue.
+	// Returns an empty slice when there is no overlap.
+	//
+	// The store does NOT exclude any specific request_id — if the caller wants to skip
+	// rows belonging to its own in-flight request (the common case when checking for
+	// duplicates of a freshly-claimed request), it should filter the returned records by
+	// RequestID itself.
 	//
-	// Liveness of the returned records' owning requests is NOT filtered here — the caller is
-	// responsible for consulting RequestStore to skip terminal owners.
+	// Liveness of the returned records' owning requests is also 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/change_store_mock.go

@@ -37,8 +37,10 @@ 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.
+// Create inserts a batch of ChangeRecords as a single multi-row INSERT IGNORE.
+// Primary-key conflicts on (queue, uri, request_id) are silently ignored so
+// queue-redelivery of the same request is a no-op. The whole batch is one
+// statement, so partial success is not observable.
 func (s *changeStore) Create(ctx context.Context, records []entity.ChangeRecord) (retErr error) {
 	op := metrics.Begin(s.scope, "create")
 	defer func() { op.Complete(retErr) }()
@@ -53,10 +55,12 @@ func (s *changeStore) Create(ctx context.Context, records []entity.ChangeRecord)
 
 	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
+		// Use the empty JSON object as the canonical "no metadata yet" value.
+		// metadata is NOT NULL in the schema, and an empty Go string would be
+		// rejected by the JSON column type.
+		metadata := r.Metadata
+		if metadata == "" {
+			metadata = "{}"
 		}
 		args = append(args, r.URI, r.RequestID, r.Queue, metadata, r.CreatedAt, r.UpdatedAt)
 	}
@@ -68,13 +72,13 @@ func (s *changeStore) Create(ctx context.Context, records []entity.ChangeRecord)
 	return nil
 }
 
-// FindOverlapping returns ChangeRecords whose uri is in the given set, scoped to queue,
-// excluding any belonging to excludeRequestID.
+// FindOverlapping returns ChangeRecords whose uri is in the given set, scoped to queue.
+// The store does not filter by request_id; callers that want to skip self should do so
+// after the call. Liveness checks against the request store are also the caller's job.
 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) }()
@@ -84,14 +88,16 @@ func (s *changeStore) FindOverlapping(
 	}
 
 	uriPlaceholders := "?" + strings.Repeat(", ?", len(uris)-1)
+	// queue leads the WHERE clause to align with the (queue, uri, request_id) PK,
+	// so this is a PK-prefix scan.
 	query := "SELECT uri, request_id, queue, metadata, created_at, updated_at FROM `change` " +
-		"WHERE uri IN (" + uriPlaceholders + ") AND queue = ? AND request_id != ?"
+		"WHERE queue = ? AND uri IN (" + uriPlaceholders + ")"
 
-	args := make([]any, 0, len(uris)+2)
+	args := make([]any, 0, 1+len(uris))
+	args = append(args, queue)
 	for _, u := range uris {
 		args = append(args, u)
 	}
-	args = append(args, queue, excludeRequestID)
 
 	rows, err := s.db.QueryContext(ctx, query, args...)
 	if err != nil {
@@ -102,13 +108,9 @@ func (s *changeStore) FindOverlapping(
 	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 {
+		if err := rows.Scan(&rec.URI, &rec.RequestID, &rec.Queue, &rec.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 {

extension/changestore/mysql/change_store.go

@@ -1,9 +1,14 @@
+-- request_id is part of the PK so concurrent claims by different requests on the
+-- same URI coexist as distinct rows. Same-request retry → PK conflict (no-op via
+-- INSERT IGNORE). Different-request collision → distinct row, surfaced by
+-- FindOverlapping. Queue leads the PK so queue-scoped lookups are PK-prefix scans
+-- and the table is shardable by queue.
 CREATE TABLE IF NOT EXISTS `change` (
     uri         VARCHAR(255) NOT NULL,
     request_id  VARCHAR(255) NOT NULL,
     queue       VARCHAR(255) NOT NULL,
-    metadata    JSON,
+    metadata    JSON NOT NULL,
     created_at  BIGINT NOT NULL,
     updated_at  BIGINT NOT NULL,
-    PRIMARY KEY (uri, request_id)
+    PRIMARY KEY (queue, uri, request_id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

extension/changestore/mysql/schema/change.sql

@@ -94,7 +94,7 @@ func (s *MySQLChangeStoreIntegrationSuite) TestCreateAndFind_NoOverlap() {
 		{URI: "github://uber/x/pull/1/aaa", RequestID: "q/1", Queue: "q", CreatedAt: 1, UpdatedAt: 1},
 	}))
 
-	got, err := s.store.FindOverlapping(s.ctx, "q", []string{"github://uber/x/pull/2/bbb"}, "q/2")
+	got, err := s.store.FindOverlapping(s.ctx, "q", []string{"github://uber/x/pull/2/bbb"})
 	require.NoError(t, err)
 	assert.Empty(t, got)
 }
@@ -106,24 +106,26 @@ func (s *MySQLChangeStoreIntegrationSuite) TestCreateAndFind_Overlap() {
 		{URI: uri, RequestID: "q/1", Queue: "q", CreatedAt: 1, UpdatedAt: 1},
 	}))
 
-	got, err := s.store.FindOverlapping(s.ctx, "q", []string{uri}, "q/2")
+	got, err := s.store.FindOverlapping(s.ctx, "q", []string{uri})
 	require.NoError(t, err)
 	require.Len(t, got, 1)
 	assert.Equal(t, "q/1", got[0].RequestID)
 	assert.Equal(t, uri, got[0].URI)
 	assert.Equal(t, "q", got[0].Queue)
 }
 
-func (s *MySQLChangeStoreIntegrationSuite) TestFindOverlapping_ExcludesSelf() {
+func (s *MySQLChangeStoreIntegrationSuite) TestFindOverlapping_ReturnsAllOwners() {
+	// The store does not exclude any specific request_id; callers filter self if they wish.
 	t := s.T()
 	uri := "github://uber/x/pull/1/aaa"
 	require.NoError(t, s.store.Create(s.ctx, []entity.ChangeRecord{
 		{URI: uri, RequestID: "q/1", Queue: "q", CreatedAt: 1, UpdatedAt: 1},
 	}))
 
-	got, err := s.store.FindOverlapping(s.ctx, "q", []string{uri}, "q/1")
+	got, err := s.store.FindOverlapping(s.ctx, "q", []string{uri})
 	require.NoError(t, err)
-	assert.Empty(t, got, "FindOverlapping must not return rows for excludeRequestID")
+	require.Len(t, got, 1, "store returns the row even when caller might consider it self")
+	assert.Equal(t, "q/1", got[0].RequestID)
 }
 
 func (s *MySQLChangeStoreIntegrationSuite) TestFindOverlapping_QueueScoped() {
@@ -133,7 +135,7 @@ func (s *MySQLChangeStoreIntegrationSuite) TestFindOverlapping_QueueScoped() {
 		{URI: uri, RequestID: "qA/1", Queue: "qA", CreatedAt: 1, UpdatedAt: 1},
 	}))
 
-	got, err := s.store.FindOverlapping(s.ctx, "qB", []string{uri}, "qB/1")
+	got, err := s.store.FindOverlapping(s.ctx, "qB", []string{uri})
 	require.NoError(t, err)
 	assert.Empty(t, got, "FindOverlapping must not return rows from a different queue")
 }
@@ -158,7 +160,7 @@ func (s *MySQLChangeStoreIntegrationSuite) TestCreate_DifferentRequestSameURI()
 		{URI: uri, RequestID: "q/2", Queue: "q", CreatedAt: 2, UpdatedAt: 2},
 	}))
 
-	got, err := s.store.FindOverlapping(s.ctx, "q", []string{uri}, "q/3")
+	got, err := s.store.FindOverlapping(s.ctx, "q", []string{uri})
 	require.NoError(t, err)
 	require.Len(t, got, 2)
 
@@ -175,12 +177,27 @@ func (s *MySQLChangeStoreIntegrationSuite) TestCreate_PreservesMetadata() {
 		{URI: uri, RequestID: "q/1", Queue: "q", Metadata: meta, CreatedAt: 1, UpdatedAt: 1},
 	}))
 
-	got, err := s.store.FindOverlapping(s.ctx, "q", []string{uri}, "q/other")
+	got, err := s.store.FindOverlapping(s.ctx, "q", []string{uri})
 	require.NoError(t, err)
 	require.Len(t, got, 1)
 	assert.JSONEq(t, meta, got[0].Metadata)
 }
 
+func (s *MySQLChangeStoreIntegrationSuite) TestCreate_EmptyMetadataStoredAsObject() {
+	// metadata is NOT NULL in the schema. The impl substitutes '{}' for an empty
+	// Metadata field so callers don't need to know about the column constraint.
+	t := s.T()
+	uri := "github://uber/x/pull/1/aaa"
+	require.NoError(t, s.store.Create(s.ctx, []entity.ChangeRecord{
+		{URI: uri, RequestID: "q/1", Queue: "q", CreatedAt: 1, UpdatedAt: 1},
+	}))
+
+	got, err := s.store.FindOverlapping(s.ctx, "q", []string{uri})
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	assert.JSONEq(t, "{}", got[0].Metadata)
+}
+
 func (s *MySQLChangeStoreIntegrationSuite) TestCreate_EmptyIsNoOp() {
 	t := s.T()
 	require.NoError(t, s.store.Create(s.ctx, nil))
@@ -196,7 +213,7 @@ func (s *MySQLChangeStoreIntegrationSuite) TestFindOverlapping_EmptyURIsIsNoOp()
 		{URI: "github://uber/x/pull/1/aaa", RequestID: "q/1", Queue: "q", CreatedAt: 1, UpdatedAt: 1},
 	}))
 
-	got, err := s.store.FindOverlapping(s.ctx, "q", nil, "q/2")
+	got, err := s.store.FindOverlapping(s.ctx, "q", nil)
 	require.NoError(t, err)
 	assert.Empty(t, got)
 }