feat(stovepipe): derive build strategy in process step

service/stovepipe/server/main.go

@@ -216,7 +216,15 @@ func run() error {
 		),
 	)
 
-	processController := process.NewController(logger.Sugar(), scope, store, queueconfigdefault.NewStore(), stovepipemq.TopicKeyProcess, "stovepipe-process")
+	processController := process.NewController(
+		logger.Sugar(),
+		scope,
+		store,
+		queueconfigdefault.NewStore(),
+		fakeSourceControlFactory{},
+		stovepipemq.TopicKeyProcess,
+		"stovepipe-process",
+	)
 	if err := primaryConsumer.Register(processController); err != nil {
 		return fmt.Errorf("failed to register process controller: %w", err)
 	}

stovepipe/controller/process/BUILD.bazel

@@ -12,6 +12,7 @@ go_library(
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/queueconfig:go_default_library",
+        "//stovepipe/extension/sourcecontrol:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
         "@org_uber_go_zap//:go_default_library",
@@ -30,6 +31,8 @@ go_test(
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/queueconfig/default:go_default_library",
+        "//stovepipe/extension/sourcecontrol:go_default_library",
+        "//stovepipe/extension/sourcecontrol/mock:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
         "//stovepipe/extension/storage/mock:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",

stovepipe/controller/process/process.go

@@ -30,6 +30,7 @@ import (
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/entity"
 	"github.com/uber/submitqueue/stovepipe/extension/queueconfig"
+	"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 	"go.uber.org/zap"
 )
@@ -42,6 +43,7 @@ type Controller struct {
 	metricsScope  tally.Scope
 	store         storage.Storage
 	queueConfigs  queueconfig.Store
+	sourceControl sourcecontrol.Factory
 	topicKey      consumer.TopicKey
 	consumerGroup string
 }
@@ -58,6 +60,7 @@ func NewController(
 	scope tally.Scope,
 	store storage.Storage,
 	queueConfigs queueconfig.Store,
+	sourceControl sourcecontrol.Factory,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
 ) *Controller {
@@ -66,6 +69,7 @@ func NewController(
 		metricsScope:  scope.SubScope("process_controller"),
 		store:         store,
 		queueConfigs:  queueConfigs,
+		sourceControl: sourceControl,
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
 	}
@@ -174,6 +178,11 @@ func (c *Controller) coalesce(ctx context.Context, request entity.Request, lates
 // re-runs coalesce-then-gate, so a slot is never spent on a now-stale head; a closed gate
 // defers (acks) rather than failing.
 func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request, queueRow entity.Queue, maxConcurrent int32) error {
+	var sc sourcecontrol.SourceControl
+	var strategy entity.BuildStrategy
+	var baseURI string
+	var err error
+
 	for {
 		if queueRow.InFlightCount >= maxConcurrent {
 			// TODO: re-enqueue the request via PublishAfter on the process topic with GateWaitDelayMs.
@@ -186,7 +195,22 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request
 			return nil
 		}
 
-		err := c.claimBuildSlot(ctx, &queueRow)
+		if queueRow.LastGreenURI != "" && sc == nil {
+			sc, err = c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue})
+			if err != nil {
+				metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1,
+					metrics.NewTag("stage", "resolve"),
+				)
+				return fmt.Errorf("ProcessController failed to resolve source control for queue %s: %w", request.Queue, err)
+			}
+		}
+
+		strategy, baseURI, err = c.deriveBuildStrategy(ctx, sc, queueRow, request)
+		if err != nil {
+			return err
+		}
+
+		err = c.claimBuildSlot(ctx, &queueRow)
 		if err == nil {
 			break
 		}
@@ -201,11 +225,7 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request
 		}
 	}
 
-	// TODO(build-strategy): derive from queue last_green_uri + SourceControl.IsAncestor.
-	request.BuildStrategy = entity.BuildStrategyFull
-	request.BaseURI = ""
-
-	transitioned, err := c.markProcessing(ctx, &request)
+	transitioned, err := c.markProcessing(ctx, &request, strategy, baseURI)
 	if err != nil {
 		// Slot claimed but never admitted: release best-effort so the slot isn't leaked
 		// (a redelivery would find the gate closed by its own claim and nothing decrements it).
@@ -220,15 +240,51 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request
 
 	// TODO(build-publish): publish BuildRequest to the build stage here.
 
+	metrics.NamedCounter(c.metricsScope, _opName, "admitted", 1,
+		metrics.NewTag("strategy", string(request.BuildStrategy)),
+	)
 	c.logger.Infow("admitted request to build",
 		"request_id", request.ID,
 		"queue", request.Queue,
 		"uri", request.URI,
 		"build_strategy", string(request.BuildStrategy),
+		"base_uri", request.BaseURI,
 	)
 	return nil
 }
 
+// deriveBuildStrategy chooses the validation scope and baseline from the queue's last-known-good commit.
+// The caller resolves source control once and persists the returned values only after successfully claiming a build slot.
+func (c *Controller) deriveBuildStrategy(ctx context.Context, sc sourcecontrol.SourceControl, queueRow entity.Queue, request entity.Request) (strategy entity.BuildStrategy, baseURI string, err error) {
+	if queueRow.LastGreenURI == "" {
+		return entity.BuildStrategyFull, "", nil
+	}
+
+	isAncestor, err := sc.IsAncestor(ctx, queueRow.LastGreenURI, request.URI)
+	if err != nil {
+		if sourcecontrol.IsNotFound(err) {
+			metrics.NamedCounter(c.metricsScope, _opName, "strategy_fallbacks", 1,
+				metrics.NewTag("reason", "unknown_ancestry"),
+			)
+			c.logger.Warnw("last-green URI is not in request history; using full build",
+				"queue", request.Queue,
+				"last_green_uri", queueRow.LastGreenURI,
+				"request_uri", request.URI,
+			)
+			return entity.BuildStrategyFull, "", nil
+		}
+		metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1,
+			metrics.NewTag("stage", "ancestry"),
+		)
+		return entity.BuildStrategyUnknown, "", fmt.Errorf("ProcessController failed to check ancestry for queue %s: %w", request.Queue, err)
+	}
+
+	if isAncestor {
+		return entity.BuildStrategyIncrementalSinceGreen, queueRow.LastGreenURI, nil
+	}
+	return entity.BuildStrategyFull, "", nil
+}
+
 // claimBuildSlot CAS-increments queue.in_flight_count by one. On version mismatch it
 // reloads queueRow and returns ErrVersionMismatch so the caller can retry.
 func (c *Controller) claimBuildSlot(ctx context.Context, queueRow *entity.Queue) error {
@@ -253,11 +309,12 @@ func (c *Controller) claimBuildSlot(ctx context.Context, queueRow *entity.Queue)
 	return nil
 }
 
-// markProcessing CAS-marks request accepted→processing, persisting BuildStrategy and BaseURI
-// already set by the admit workflow. Retries on version conflicts. transitioned is true only
-// when this call performed the CAS; false means a concurrent writer already advanced the
-// request past accepted (a lost admit race), so the caller must release its claimed slot.
-func (c *Controller) markProcessing(ctx context.Context, request *entity.Request) (transitioned bool, err error) {
+// markProcessing CAS-marks request accepted→processing and persists the strategy chosen after the
+// build slot was claimed. It reapplies the values after a request reload so an accepted concurrent
+// update cannot discard them. transitioned is true only when this call performed the CAS; false
+// means a concurrent writer already advanced the request past accepted, so the caller must release
+// its claimed slot.
+func (c *Controller) markProcessing(ctx context.Context, request *entity.Request, strategy entity.BuildStrategy, baseURI string) (transitioned bool, err error) {
 	reqStore := c.store.GetRequestStore()
 
 	for {
@@ -267,6 +324,8 @@ func (c *Controller) markProcessing(ctx context.Context, request *entity.Request
 
 		updated := *request
 		updated.State = entity.RequestStateProcessing
+		updated.BuildStrategy = strategy
+		updated.BaseURI = baseURI
 		newVersion := request.Version + 1
 		if err := reqStore.Update(ctx, updated, request.Version, newVersion); err != nil {
 			if errors.Is(err, storage.ErrVersionMismatch) {