validate: per-URI loop calls GetByURI with early termination

Adapts to the per-record / per-URI ChangeStore API:
- start.claimURIs loops Create per URI; each call is independent and
  idempotent on PK conflict, so partial-failure retries converge.
- validate.checkDuplicate iterates request URIs, calls GetByURI for each,
  filters out self, dedupes seen owners, and short-circuits on the first
  live duplicate.
- Test cases restructured to map records by URI (since the controller now
  queries one URI at a time) and to exercise multi-URI requests.

Author: behinddwalls

orchestrator/controller/start/start.go

@@ -148,26 +148,26 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	return nil // Success - message will be acked
 }
 
-// claimURIs persists one ChangeRecord per URI in the request. The change store's
-// INSERT IGNORE semantics make this idempotent on queue redelivery (same (URI, RequestID)
-// is a no-op). Different requests with overlapping URIs do NOT collide on insert; the
-// validate controller queries the change store to detect that overlap.
+// claimURIs persists one ChangeRecord per URI in the request. Each Create call is
+// independent; the change store's per-PK idempotency makes the loop safe under
+// queue redelivery (same (Queue, URI, RequestID) is a no-op on retry). Different
+// requests with overlapping URIs do NOT collide on insert; the validate controller
+// queries the change store to detect that overlap.
 func (c *Controller) claimURIs(ctx context.Context, request entity.Request) error {
-	if len(request.Change.URIs) == 0 {
-		return nil
-	}
 	now := time.Now().UnixMilli()
-	records := make([]entity.ChangeRecord, 0, len(request.Change.URIs))
 	for _, uri := range request.Change.URIs {
-		records = append(records, entity.ChangeRecord{
+		record := entity.ChangeRecord{
 			URI:       uri,
 			RequestID: request.ID,
 			Queue:     request.Queue,
 			CreatedAt: now,
 			UpdatedAt: now,
-		})
+		}
+		if err := c.changeStore.Create(ctx, record); err != nil {
+			return fmt.Errorf("failed to claim uri=%s for request %s: %w", uri, request.ID, err)
+		}
 	}
-	return c.changeStore.Create(ctx, records)
+	return nil
 }
 
 // publish publishes a request ID to the specified topic key.

orchestrator/controller/start/start_test.go

@@ -200,11 +200,11 @@ func TestController_Process_MultipleChanges(t *testing.T) {
 	cs := changemock.NewMockChangeStore(ctrl)
 	var captured []entity.ChangeRecord
 	cs.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(
-		func(ctx context.Context, records []entity.ChangeRecord) error {
-			captured = records
+		func(ctx context.Context, record entity.ChangeRecord) error {
+			captured = append(captured, record)
 			return nil
 		},
-	)
+	).Times(3)
 
 	controller := newTestController(t, ctrl, newMockStorage(ctrl), cs, nil)
 

orchestrator/controller/validate/validate.go

@@ -165,47 +165,42 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	return nil // Success - message will be acked
 }
 
-// checkDuplicate queries the change store for any other in-flight request whose URIs
-// overlap with this request's. The change rows themselves are written upstream by the
-// start controller; validate is read-only here. For each unique candidate request_id
-// returned, this consults the request store to skip terminal owners and orphans
-// (ErrNotFound). Returns the first live-duplicate request_id found, or "" if none.
+// checkDuplicate looks for any other in-flight request whose URIs overlap with this
+// request's. The change rows themselves are written upstream by the start controller;
+// validate is read-only here. For each URI it queries the change store, walks the
+// returned candidates skipping self/duplicates/orphans/terminals, and short-circuits
+// on the first live duplicate. Returns that request_id, or "" if none.
 //
-// Two single-table queries by design — no cross-store SQL joins.
+// Per-URI / per-record reads keep the contract backend-agnostic; the typical request
+// has 1-5 URIs, so the loop is cheap.
 func (c *Controller) checkDuplicate(ctx context.Context, request entity.Request) (string, error) {
-	if len(request.Change.URIs) == 0 {
-		return "", nil
-	}
-
-	overlaps, err := c.changeStore.FindOverlapping(ctx, request.Queue, request.Change.URIs)
-	if err != nil {
-		coremetrics.NamedCounter(c.metricsScope, "process", "change_store_query_errors", 1)
-		return "", fmt.Errorf("failed to query overlapping changes for request %s: %w", request.ID, err)
-	}
-
-	// Liveness check: an overlap is only a duplicate if the owning request is non-terminal.
-	// The store does not exclude self, so skip our own request_id here. Walk unique candidate
-	// request_ids; orphans (ErrNotFound) and terminal owners are skipped.
-	seen := make(map[string]struct{}, len(overlaps))
-	for _, rec := range overlaps {
-		if rec.RequestID == request.ID {
-			continue // skip rows belonging to this request itself
-		}
-		if _, ok := seen[rec.RequestID]; ok {
-			continue
-		}
-		seen[rec.RequestID] = struct{}{}
-
-		owner, err := c.store.GetRequestStore().Get(ctx, rec.RequestID)
-		if errors.Is(err, storage.ErrNotFound) {
-			continue
-		}
+	seenOwners := make(map[string]struct{})
+	for _, uri := range request.Change.URIs {
+		records, err := c.changeStore.GetByURI(ctx, request.Queue, uri)
 		if err != nil {
-			coremetrics.NamedCounter(c.metricsScope, "process", "storage_errors", 1)
-			return "", fmt.Errorf("failed to look up overlapping request %s: %w", rec.RequestID, err)
+			coremetrics.NamedCounter(c.metricsScope, "process", "change_store_query_errors", 1)
+			return "", fmt.Errorf("failed to query change store for request %s uri=%s: %w", request.ID, uri, err)
 		}
-		if !entity.IsRequestStateTerminal(owner.State) {
-			return rec.RequestID, nil
+		for _, rec := range records {
+			if rec.RequestID == request.ID {
+				continue // skip rows belonging to this request itself
+			}
+			if _, ok := seenOwners[rec.RequestID]; ok {
+				continue
+			}
+			seenOwners[rec.RequestID] = struct{}{}
+
+			owner, err := c.store.GetRequestStore().Get(ctx, rec.RequestID)
+			if errors.Is(err, storage.ErrNotFound) {
+				continue
+			}
+			if err != nil {
+				coremetrics.NamedCounter(c.metricsScope, "process", "storage_errors", 1)
+				return "", fmt.Errorf("failed to look up overlapping request %s: %w", rec.RequestID, err)
+			}
+			if !entity.IsRequestStateTerminal(owner.State) {
+				return rec.RequestID, nil
+			}
 		}
 	}
 	return "", nil

orchestrator/controller/validate/validate_test.go

@@ -81,11 +81,11 @@ func newMockStorage(ctrl *gomock.Controller, request entity.Request) (*storagemo
 }
 
 // newMockChangeStore creates a MockChangeStore with default no-overlap behavior.
-// Tests that need to simulate overlap can override FindOverlapping with their own EXPECT.
+// Tests that need to simulate overlap can override GetByURI with their own EXPECT.
 // Validate is read-only against the change store — it never calls Create.
 func newMockChangeStore(ctrl *gomock.Controller) *changemock.MockChangeStore {
 	cs := changemock.NewMockChangeStore(ctrl)
-	cs.EXPECT().FindOverlapping(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
+	cs.EXPECT().GetByURI(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
 	return cs
 }
 
@@ -278,61 +278,71 @@ func TestController_Process_DuplicateDetection(t *testing.T) {
 		queueName     = "test-queue"
 		newRequestID  = queueName + "/123"
 		dupRequestID  = queueName + "/100"
-		uri           = "github://uber/service/pull/1/abc"
+		uriA          = "github://uber/service/pull/1/abc"
+		uriB          = "github://uber/service/pull/2/def"
 		anotherReqID  = queueName + "/050"
 		orphanReqID   = queueName + "/999"
 		terminalReqID = queueName + "/200"
 	)
 
 	tests := []struct {
 		name           string
-		overlap        []entity.ChangeRecord
+		requestURIs    []string                         // URIs on the new request; defaults to [uriA]
+		byURI          map[string][]entity.ChangeRecord // GetByURI mock returns
 		ownerLookup    map[string]entity.Request
 		ownerNotFound  map[string]bool
 		ownerErr       map[string]error
 		wantUserErr    bool
 		wantUnexpected bool
 	}{
 		{
-			name:    "no overlap proceeds to merge check",
-			overlap: nil,
+			name:  "no overlap proceeds to merge check",
+			byURI: map[string][]entity.ChangeRecord{uriA: nil},
 		},
 		{
-			name:    "overlap with live in-flight request returns user error",
-			overlap: []entity.ChangeRecord{{URI: uri, RequestID: dupRequestID, Queue: queueName}},
+			name: "overlap with live in-flight request returns user error",
+			byURI: map[string][]entity.ChangeRecord{
+				uriA: {{URI: uriA, RequestID: dupRequestID, Queue: queueName}},
+			},
 			ownerLookup: map[string]entity.Request{
 				dupRequestID: {ID: dupRequestID, Queue: queueName, State: entity.RequestStateStarted, Version: 1},
 			},
 			wantUserErr: true,
 		},
 		{
-			name:    "overlap with terminal owner is skipped",
-			overlap: []entity.ChangeRecord{{URI: uri, RequestID: terminalReqID, Queue: queueName}},
+			name: "overlap with terminal owner is skipped",
+			byURI: map[string][]entity.ChangeRecord{
+				uriA: {{URI: uriA, RequestID: terminalReqID, Queue: queueName}},
+			},
 			ownerLookup: map[string]entity.Request{
 				terminalReqID: {ID: terminalReqID, Queue: queueName, State: entity.RequestStateLanded, Version: 5},
 			},
 		},
 		{
-			name:          "overlap with orphan owner (ErrNotFound) is skipped",
-			overlap:       []entity.ChangeRecord{{URI: uri, RequestID: orphanReqID, Queue: queueName}},
+			name: "overlap with orphan owner (ErrNotFound) is skipped",
+			byURI: map[string][]entity.ChangeRecord{
+				uriA: {{URI: uriA, RequestID: orphanReqID, Queue: queueName}},
+			},
 			ownerNotFound: map[string]bool{orphanReqID: true},
 		},
 		{
-			name: "multiple URIs same owner deduped to single Get call",
-			overlap: []entity.ChangeRecord{
-				{URI: uri, RequestID: dupRequestID, Queue: queueName},
-				{URI: "github://uber/service/pull/2/def", RequestID: dupRequestID, Queue: queueName},
+			name:        "multi-URI same owner deduped to single Get call",
+			requestURIs: []string{uriA, uriB},
+			byURI: map[string][]entity.ChangeRecord{
+				uriA: {{URI: uriA, RequestID: dupRequestID, Queue: queueName}},
+				uriB: {{URI: uriB, RequestID: dupRequestID, Queue: queueName}},
 			},
 			ownerLookup: map[string]entity.Request{
 				dupRequestID: {ID: dupRequestID, Queue: queueName, State: entity.RequestStateValidated, Version: 2},
 			},
 			wantUserErr: true,
 		},
 		{
-			name: "first owner terminal then second live picks the live one",
-			overlap: []entity.ChangeRecord{
-				{URI: uri, RequestID: terminalReqID, Queue: queueName},
-				{URI: "github://uber/service/pull/2/def", RequestID: anotherReqID, Queue: queueName},
+			name:        "first URI's owner is terminal, second URI's owner is live",
+			requestURIs: []string{uriA, uriB},
+			byURI: map[string][]entity.ChangeRecord{
+				uriA: {{URI: uriA, RequestID: terminalReqID, Queue: queueName}},
+				uriB: {{URI: uriB, RequestID: anotherReqID, Queue: queueName}},
 			},
 			ownerLookup: map[string]entity.Request{
 				terminalReqID: {ID: terminalReqID, State: entity.RequestStateError, Version: 3},
@@ -342,25 +352,29 @@ func TestController_Process_DuplicateDetection(t *testing.T) {
 		},
 		{
 			// Store doesn't exclude self; controller filters by RequestID and must not look up its own row.
-			name: "self row in overlap is filtered (no Get call)",
-			overlap: []entity.ChangeRecord{
-				{URI: uri, RequestID: newRequestID, Queue: queueName},
+			name: "self row in result is filtered (no Get call)",
+			byURI: map[string][]entity.ChangeRecord{
+				uriA: {{URI: uriA, RequestID: newRequestID, Queue: queueName}},
 			},
 		},
 		{
 			name: "self row mixed with live other returns the other",
-			overlap: []entity.ChangeRecord{
-				{URI: uri, RequestID: newRequestID, Queue: queueName},
-				{URI: uri, RequestID: dupRequestID, Queue: queueName},
+			byURI: map[string][]entity.ChangeRecord{
+				uriA: {
+					{URI: uriA, RequestID: newRequestID, Queue: queueName},
+					{URI: uriA, RequestID: dupRequestID, Queue: queueName},
+				},
 			},
 			ownerLookup: map[string]entity.Request{
 				dupRequestID: {ID: dupRequestID, Queue: queueName, State: entity.RequestStateStarted, Version: 1},
 			},
 			wantUserErr: true,
 		},
 		{
-			name:    "owner lookup unexpected error propagates",
-			overlap: []entity.ChangeRecord{{URI: uri, RequestID: dupRequestID, Queue: queueName}},
+			name: "owner lookup unexpected error propagates",
+			byURI: map[string][]entity.ChangeRecord{
+				uriA: {{URI: uriA, RequestID: dupRequestID, Queue: queueName}},
+			},
 			ownerErr: map[string]error{
 				dupRequestID: fmt.Errorf("db down"),
 			},
@@ -373,10 +387,15 @@ func TestController_Process_DuplicateDetection(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			mc := newMergeableMock(ctrl)
 
+			uris := tt.requestURIs
+			if uris == nil {
+				uris = []string{uriA}
+			}
+
 			request := entity.Request{
 				ID:           newRequestID,
 				Queue:        queueName,
-				Change:       entity.Change{URIs: []string{uri}},
+				Change:       entity.Change{URIs: uris},
 				LandStrategy: entity.RequestLandStrategyRebase,
 				State:        entity.RequestStateStarted,
 				Version:      1,
@@ -397,7 +416,11 @@ func TestController_Process_DuplicateDetection(t *testing.T) {
 			store.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes()
 
 			cs := changemock.NewMockChangeStore(ctrl)
-			cs.EXPECT().FindOverlapping(gomock.Any(), queueName, []string{uri}).Return(tt.overlap, nil)
+			// One GetByURI per URI on the request, in order. Controller short-circuits on first
+			// live duplicate, so .AnyTimes() lets unmatched URIs go un-queried.
+			for _, u := range uris {
+				cs.EXPECT().GetByURI(gomock.Any(), queueName, u).Return(tt.byURI[u], nil).MaxTimes(1)
+			}
 
 			controller := newTestController(t, ctrl, store, cs, mc, nil)
 
@@ -437,7 +460,7 @@ func TestController_Process_ChangeStoreQueryFailure(t *testing.T) {
 	store, _ := newMockStorage(ctrl, request)
 
 	cs := changemock.NewMockChangeStore(ctrl)
-	cs.EXPECT().FindOverlapping(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("change store down"))
+	cs.EXPECT().GetByURI(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("change store down"))
 
 	controller := newTestController(t, ctrl, store, cs, mc, nil)