move URI claim back to start

start writes the change rows on entry; validate stays read-only and uses the
change store + request store liveness check to detect duplicates. Trade-off:
validation-failing requests leave behind change rows whose owner request will
become terminal, which the controller-side liveness check filters out at
duplicate-detection time.

Reverses the previous move-claim-to-validate commit. The earlier-claim story
trades simpler validate (read-only) for orphan claim rows (filtered, not
blocking).

Author: behinddwalls

example/server/orchestrator/main.go

@@ -406,6 +406,7 @@ func registerControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope t
 		logger,
 		scope,
 		store,
+		changeStore,
 		registry,
 		consumer.TopicKeyStart,
 		"orchestrator-start",

orchestrator/controller/start/BUILD.bazel

@@ -10,6 +10,7 @@ go_library(
         "//core/request",
         "//entity",
         "//entity/queue",
+        "//extension/changestore",
         "//extension/storage",
         "@com_github_uber_go_tally_v4//:tally",
         "@org_uber_go_zap//:zap",
@@ -25,6 +26,7 @@ go_test(
         "//core/errs",
         "//entity",
         "//entity/queue",
+        "//extension/changestore/mock",
         "//extension/queue/mock",
         "//extension/storage",
         "//extension/storage/mock",

orchestrator/controller/start/start.go

@@ -18,24 +18,29 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/uber-go/tally/v4"
 	"github.com/uber/submitqueue/core/consumer"
 	corerequest "github.com/uber/submitqueue/core/request"
 	"github.com/uber/submitqueue/entity"
 	entityqueue "github.com/uber/submitqueue/entity/queue"
+	"github.com/uber/submitqueue/extension/changestore"
 	"github.com/uber/submitqueue/extension/storage"
 	"go.uber.org/zap"
 )
 
 // Controller handles start queue messages.
-// It consumes requests, persists them to storage, and publishes to the validate stage.
-// Duplicate detection and URI claiming happen downstream in the validate controller.
-// Implements consumer.Controller.
+// It consumes requests, persists them to the request store, claims their URIs in
+// the change store, and publishes to the validate stage. Both writes are idempotent
+// on retries; the duplicate-detection check itself is performed downstream by the
+// validate controller, which reads the change store and consults the request store
+// for liveness. Implements consumer.Controller.
 type Controller struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
 	store         storage.Storage
+	changeStore   changestore.ChangeStore
 	registry      consumer.TopicRegistry
 	topicKey      consumer.TopicKey
 	consumerGroup string
@@ -49,6 +54,7 @@ func NewController(
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
 	store storage.Storage,
+	changeStore changestore.ChangeStore,
 	registry consumer.TopicRegistry,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
@@ -57,29 +63,28 @@ func NewController(
 		logger:        logger.Named("start_controller"),
 		metricsScope:  scope.SubScope("start_controller"),
 		store:         store,
+		changeStore:   changeStore,
 		registry:      registry,
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
 	}
 }
 
 // Process processes a request delivery from the queue.
-// Deserializes the request, persists it, and publishes to the validate topic.
+// Persists the request, claims its URIs in the change store, and publishes to validate.
 // Returns nil to ack (success), or error to nack (retry).
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error {
 	c.metricsScope.Counter("received").Inc(1)
 
 	msg := delivery.Message()
 
-	// Deserialize land request from gateway
 	landRequest, err := entity.LandRequestFromBytes(msg.Payload)
 	if err != nil {
 		c.metricsScope.Counter("deserialize_errors").Inc(1)
 		// Non-retryable: malformed messages will never succeed regardless of retry count
 		return fmt.Errorf("failed to deserialize land request: %w", err)
 	}
 
-	// Construct the full versioned Request entity with orchestrator-owned fields
 	request := entity.Request{
 		ID:           landRequest.ID,
 		Queue:        landRequest.Queue,
@@ -103,22 +108,27 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 
 	// Persist request to storage. ErrAlreadyExists means a queue redelivery of the same
 	// request_id (an at-least-once retry of THIS message), not a cross-request collision.
-	// Cross-request URI overlap is detected and claims are written downstream in validate.
 	if err := c.store.GetRequestStore().Create(ctx, request); err != nil && !errors.Is(err, storage.ErrAlreadyExists) {
 		c.metricsScope.Counter("storage_errors").Inc(1)
 		return fmt.Errorf("failed to create request: %w", err)
 	}
 
-	// Record the "new" status in the request log
+	// Claim each URI in the change store. Different requests on the same URI write
+	// distinct rows (different request_id), so cross-request URI overlap does not
+	// collide on insert; the validate controller surfaces it via GetByURI + a
+	// liveness check against the request store.
+	if err := c.claimURIs(ctx, request); err != nil {
+		c.metricsScope.Counter("change_store_errors").Inc(1)
+		return fmt.Errorf("failed to claim URIs for request %s: %w", request.ID, err)
+	}
+
+	// Record the "new" status in the request log.
 	logEntry := entity.NewRequestLog(request.ID, entity.RequestStatusStarted, request.Version, "", nil)
-	// Using request.ID as the partition key to ensure ordering of log entries for the same request
-	// and parallel processing of log entries for different requests.
 	if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID); err != nil {
 		c.metricsScope.Counter("request_log_errors").Inc(1)
 		return fmt.Errorf("failed to publish request log: %w", err)
 	}
 
-	// Publish to validate topic
 	if err := c.publish(ctx, consumer.TopicKeyValidate, request.ID, request.Queue); err != nil {
 		c.metricsScope.Counter("publish_errors").Inc(1)
 		return fmt.Errorf("failed to publish to validate: %w", err)
@@ -130,8 +140,28 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	)
 
 	c.metricsScope.Counter("processed").Inc(1)
+	return nil
+}
 
-	return nil // Success - message will be acked
+// 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).
+func (c *Controller) claimURIs(ctx context.Context, request entity.Request) error {
+	now := time.Now().UnixMilli()
+	for _, uri := range request.Change.URIs {
+		record := entity.ChangeRecord{
+			URI:       uri,
+			RequestID: request.ID,
+			Queue:     request.Queue,
+			CreatedAt: now,
+			UpdatedAt: now,
+			Version:   1,
+		}
+		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 nil
 }
 
 // publish publishes a request ID to the specified topic key.

orchestrator/controller/start/start_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/uber/submitqueue/core/errs"
 	"github.com/uber/submitqueue/entity"
 	"github.com/uber/submitqueue/entity/queue"
+	changemock "github.com/uber/submitqueue/extension/changestore/mock"
 	queuemock "github.com/uber/submitqueue/extension/queue/mock"
 	"github.com/uber/submitqueue/extension/storage"
 	storagemock "github.com/uber/submitqueue/extension/storage/mock"
@@ -34,7 +35,13 @@ import (
 )
 
 // newTestController creates a controller with test dependencies.
-func newTestController(t *testing.T, ctrl *gomock.Controller, store *storagemock.MockStorage, publishErr error) *Controller {
+func newTestController(
+	t *testing.T,
+	ctrl *gomock.Controller,
+	store *storagemock.MockStorage,
+	cs *changemock.MockChangeStore,
+	publishErr error,
+) *Controller {
 	logger := zaptest.NewLogger(t).Sugar()
 	scope := tally.NoopScope
 
@@ -56,7 +63,7 @@ func newTestController(t *testing.T, ctrl *gomock.Controller, store *storagemock
 	)
 	require.NoError(t, err)
 
-	return NewController(logger, scope, store, registry, consumer.TopicKeyStart, "orchestrator-start")
+	return NewController(logger, scope, store, cs, registry, consumer.TopicKeyStart, "orchestrator-start")
 }
 
 // newMockStorage creates a MockStorage with a MockRequestStore that succeeds on Create.
@@ -69,6 +76,13 @@ func newMockStorage(ctrl *gomock.Controller) *storagemock.MockStorage {
 	return store
 }
 
+// newMockChangeStore returns a MockChangeStore that accepts any Create call.
+func newMockChangeStore(ctrl *gomock.Controller) *changemock.MockChangeStore {
+	cs := changemock.NewMockChangeStore(ctrl)
+	cs.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	return cs
+}
+
 // makeDelivery builds a MockDelivery wrapping a serialized LandRequest.
 func makeDelivery(t *testing.T, ctrl *gomock.Controller, lr entity.LandRequest) *queuemock.MockDelivery {
 	payload, err := lr.ToBytes()
@@ -83,7 +97,7 @@ func makeDelivery(t *testing.T, ctrl *gomock.Controller, lr entity.LandRequest)
 
 func TestNewController(t *testing.T) {
 	ctrl := gomock.NewController(t)
-	controller := newTestController(t, ctrl, newMockStorage(ctrl), nil)
+	controller := newTestController(t, ctrl, newMockStorage(ctrl), newMockChangeStore(ctrl), nil)
 
 	require.NotNil(t, controller)
 	assert.Equal(t, consumer.TopicKeyStart, controller.TopicKey())
@@ -93,7 +107,7 @@ func TestNewController(t *testing.T) {
 
 func TestController_Process_Success(t *testing.T) {
 	ctrl := gomock.NewController(t)
-	controller := newTestController(t, ctrl, newMockStorage(ctrl), nil)
+	controller := newTestController(t, ctrl, newMockStorage(ctrl), newMockChangeStore(ctrl), nil)
 
 	delivery := makeDelivery(t, ctrl, entity.LandRequest{
 		ID:           "test-queue/123",
@@ -107,7 +121,7 @@ func TestController_Process_Success(t *testing.T) {
 
 func TestController_Process_InvalidJSON(t *testing.T) {
 	ctrl := gomock.NewController(t)
-	controller := newTestController(t, ctrl, newMockStorage(ctrl), nil)
+	controller := newTestController(t, ctrl, newMockStorage(ctrl), newMockChangeStore(ctrl), nil)
 
 	invalidPayload := []byte(`{"invalid": json"}`)
 	msg := queue.NewMessage("invalid-msg", invalidPayload, "partition1", nil)
@@ -134,7 +148,7 @@ func TestController_Process_ConstructsRequestWithStateAndVersion(t *testing.T) {
 	store := storagemock.NewMockStorage(ctrl)
 	store.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes()
 
-	controller := newTestController(t, ctrl, store, nil)
+	controller := newTestController(t, ctrl, store, newMockChangeStore(ctrl), nil)
 
 	landRequest := entity.LandRequest{
 		ID:           "test-queue/42",
@@ -166,7 +180,7 @@ func TestController_Process_AllStrategies(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			ctrl := gomock.NewController(t)
-			controller := newTestController(t, ctrl, newMockStorage(ctrl), nil)
+			controller := newTestController(t, ctrl, newMockStorage(ctrl), newMockChangeStore(ctrl), nil)
 
 			delivery := makeDelivery(t, ctrl, entity.LandRequest{
 				ID:           fmt.Sprintf("queue/%s", tt.strategy),
@@ -182,27 +196,46 @@ func TestController_Process_AllStrategies(t *testing.T) {
 
 func TestController_Process_MultipleChanges(t *testing.T) {
 	ctrl := gomock.NewController(t)
-	controller := newTestController(t, ctrl, newMockStorage(ctrl), nil)
 
-	delivery := makeDelivery(t, ctrl, entity.LandRequest{
-		ID:    "queue/999",
-		Queue: "test-queue",
-		Change: entity.Change{
-			URIs: []string{
-				"github://uber/monorepo/pull/1/aaa111",
-				"github://uber/monorepo/pull/2/bbb222",
-				"github://uber/monorepo/pull/3/ccc333",
-			},
+	cs := changemock.NewMockChangeStore(ctrl)
+	var captured []entity.ChangeRecord
+	cs.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, record entity.ChangeRecord) error {
+			captured = append(captured, record)
+			return nil
 		},
+	).Times(3)
+
+	controller := newTestController(t, ctrl, newMockStorage(ctrl), cs, nil)
+
+	uris := []string{
+		"github://uber/monorepo/pull/1/aaa111",
+		"github://uber/monorepo/pull/2/bbb222",
+		"github://uber/monorepo/pull/3/ccc333",
+	}
+	delivery := makeDelivery(t, ctrl, entity.LandRequest{
+		ID:           "queue/999",
+		Queue:        "test-queue",
+		Change:       entity.Change{URIs: uris},
 		LandStrategy: entity.RequestLandStrategySquashRebase,
 	})
 
 	require.NoError(t, controller.Process(context.Background(), delivery))
+
+	require.Len(t, captured, len(uris))
+	for i, r := range captured {
+		assert.Equal(t, uris[i], r.URI)
+		assert.Equal(t, "queue/999", r.RequestID)
+		assert.Equal(t, "test-queue", r.Queue)
+		assert.Equal(t, int32(1), r.Version)
+		assert.Positive(t, r.CreatedAt)
+		assert.Equal(t, r.CreatedAt, r.UpdatedAt)
+	}
 }
 
 func TestController_Process_PublishFailure(t *testing.T) {
 	ctrl := gomock.NewController(t)
-	controller := newTestController(t, ctrl, newMockStorage(ctrl), fmt.Errorf("publish failed"))
+	controller := newTestController(t, ctrl, newMockStorage(ctrl), newMockChangeStore(ctrl), fmt.Errorf("publish failed"))
 
 	delivery := makeDelivery(t, ctrl, entity.LandRequest{
 		ID:           "test-queue/123",
@@ -222,7 +255,7 @@ func TestController_Process_StorageFailure(t *testing.T) {
 	store := storagemock.NewMockStorage(ctrl)
 	store.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes()
 
-	controller := newTestController(t, ctrl, store, nil)
+	controller := newTestController(t, ctrl, store, newMockChangeStore(ctrl), nil)
 
 	delivery := makeDelivery(t, ctrl, entity.LandRequest{
 		ID:           "test-queue/123",
@@ -244,7 +277,7 @@ func TestController_Process_AlreadyExistsSucceeds(t *testing.T) {
 	store := storagemock.NewMockStorage(ctrl)
 	store.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes()
 
-	controller := newTestController(t, ctrl, store, nil)
+	controller := newTestController(t, ctrl, store, newMockChangeStore(ctrl), nil)
 
 	delivery := makeDelivery(t, ctrl, entity.LandRequest{
 		ID:           "test-queue/123",
@@ -256,9 +289,27 @@ func TestController_Process_AlreadyExistsSucceeds(t *testing.T) {
 	require.NoError(t, controller.Process(context.Background(), delivery))
 }
 
+func TestController_Process_ChangeStoreFailure(t *testing.T) {
+	ctrl := gomock.NewController(t)
+
+	cs := changemock.NewMockChangeStore(ctrl)
+	cs.EXPECT().Create(gomock.Any(), gomock.Any()).Return(fmt.Errorf("change store down"))
+
+	controller := newTestController(t, ctrl, newMockStorage(ctrl), cs, nil)
+
+	delivery := makeDelivery(t, ctrl, entity.LandRequest{
+		ID:           "test-queue/123",
+		Queue:        "test-queue",
+		Change:       entity.Change{URIs: []string{"github://uber/service/pull/1/xyz789abc"}},
+		LandStrategy: entity.RequestLandStrategyRebase,
+	})
+
+	require.Error(t, controller.Process(context.Background(), delivery))
+}
+
 func TestController_InterfaceImplementation(t *testing.T) {
 	ctrl := gomock.NewController(t)
-	controller := newTestController(t, ctrl, newMockStorage(ctrl), nil)
+	controller := newTestController(t, ctrl, newMockStorage(ctrl), newMockChangeStore(ctrl), nil)
 
 	var _ consumer.Controller = controller
 }