feat(stovepipe): Promote green commits to a ref

Author: prathshenoy

doc/rfc/stovepipe/workflow.md

@@ -43,11 +43,17 @@ Greenness is recorded as a **health degree** where **`0` means green** and **hig
 
 A **project** is a caller-defined slice of the repository. Whole-repo greenness answers "is the branch green at this URI"; project greenness answers the question deployments actually need — **"is *this project* green at this URI"**, and its dual, "what is the latest URI at which this project is green". Projects are derived from the build's **target graph**: analysis sees which targets broke and maps them to projects. How targets map to projects is implementer-specific (directory ownership, build metadata, an external service) and lives behind the project-analysis stage, not in the core pipeline.
 
+### Promotion ref — the last green commit, by name
+
+A Queue may have a **promotion ref**: a stable branch name (say `verified-main` for `monorepo/main`) that Stovepipe advances to each commit it establishes green. A deploy gate or cache warmer then fetches that name and needs to know nothing about Stovepipe, URIs, or greenness degrees. It is the pull-shaped counterpart to Hooks' push: the same fact, available to consumers that would rather resolve a ref than subscribe to an event.
+
+The ref is a *cache* of the last-green URI, not a second record of greenness. It only ever moves where the bookmark already points, so it inherits the same forward-only rule, and a commit that a history rewrite has dropped from the branch is skipped rather than retried — the next green commit corrects the ref. Which ref a Queue promotes to, and whether it has one at all, is `SourceControl` configuration resolved from the Queue name alongside the repo and credentials; the pipeline names only the commit, never a branch.
+
 ## Extensions
 
 | Extension | Responsibility |
 |---|---|
-| **SourceControl** | Resolve a Queue name to its current head URI; answer ancestry/comparison questions between two URIs (is the new head a fast-forward descendant of the last green, or was history rewritten?); enumerate commits in a range. The sole owner of URI semantics. |
+| **SourceControl** | Resolve a Queue name to its current head URI; answer ancestry/comparison questions between two URIs (is the new head a fast-forward descendant of the last green, or was history rewritten?); enumerate commits in a range; advance the Queue's **promotion ref** to a commit. The sole owner of URI semantics, including which refs a Queue name resolves to. |
 | **build-runner** | Build a scope at a URI (optionally relative to a baseline URI), returning pass/fail and the target graph. See [build-runner.md](../submitqueue/build-runner.md). |
 | **Hooks** | Publish Stovepipe's greenness events to downstream systems — "this URI / this project is now green (or not green)". Fire-and-forget notification, decoupled so Stovepipe does not know or care who consumes the event. |
 | **Storage** | Persist Queues (incl. last-green URI), Requests, build records, and per-URI / per-project greenness. Key/value-shaped per the extension-design rules in [CLAUDE.md](../../../CLAUDE.md). |
@@ -129,7 +135,7 @@ The pipeline runs in two phases against the same Request. **Phase 1** establishe
 2. **process** — decides build strategy (incremental since last-green vs full monorepo), gates concurrent work per Queue, coalesces backlog to the latest head, and publishes to `build`. See [process.md](steps/process.md).
 3. **build** — runs the build-runner for the chosen scope. A flag derived from `process` decides whether to build relative to the last-green **baseline URI** (incremental) or from scratch (full). It records a build and publishes the BuildID.
 4. **buildsignal** — records the build's status and target graph when the build completes, then releases the Queue's `in_flight_count` slot, projects the terminal status onto the Request (`succeeded` / `failed` / `cancelled`), and publishes the RequestID to `record`.
-5. **record** — writes the whole-repo greenness for the head URI (`0` green / `1` broken to start), derived from the Request's build outcome. On green it advances the Queue's **last-green URI** so the next `process` can build incrementally from here. It fires the **Hooks** extension with the green/not-green event, then fans out into Phase 2. The Queue's `in_flight_count` was already released by `buildsignal` when the build went terminal.
+5. **record** — writes the whole-repo greenness for the head URI (`0` green / `1` broken to start), derived from the Request's build outcome. On green it advances the Queue's **last-green URI** so the next `process` can build incrementally from here, and asks `SourceControl` to advance the Queue's **promotion ref** to the same commit (see [Promotion ref](#promotion-ref)). It fires the **Hooks** extension with the green/not-green event, then fans out into Phase 2. The Queue's `in_flight_count` was already released by `buildsignal` when the build went terminal.
 
 ### Phase 2 — project greenness
 
@@ -147,7 +153,7 @@ The pipeline runs in two phases against the same Request. **Phase 1** establishe
 | **process** | RequestID | build | Build strategy, concurrency gate, backlog coalescing → [process.md](steps/process.md) |
 | **build** | RequestID | buildsignal | Run the build-runner for the chosen scope; baseline = last-green URI iff incremental |
 | **buildsignal** | BuildID | record (P1), record (P2) | Record build status + target graph; release `in_flight_count`; project the outcome onto the Request; signal completion |
-| **record** | RequestID | analyze (P1→P2), Hooks | Write greenness; advance last-green URI on whole-repo green; fire Hooks |
+| **record** | RequestID | analyze (P1→P2), Hooks | Write greenness; on whole-repo green advance last-green URI and the promotion ref; fire Hooks |
 | **analyze** | RequestID | build | Map broken/at-risk targets → projects; decide project-scoped builds |
 
 ## Step RFCs

service/stovepipe/README.md

@@ -15,7 +15,7 @@ Stovepipe therefore needs two MySQL databases: a **storage** database (the `requ
 `server/main.go` is the composition root and supplies the concrete extension implementations. Two are deliberately demo-only and must be replaced for any real deployment:
 
 - **`inMemoryCounter`** — a process-local `counter.Counter` for sequence numbers; not durable. A real deployment uses a persistent implementation (e.g. `platform/extension/counter/mysql`).
-- **`fakeSourceControlFactory`** — seeds each queue with a deterministic single-commit history so ingest resolves a stable head URI (and re-ingesting the same queue exercises the dedup path). A real deployment supplies a VCS-backed `sourcecontrol.Factory`.
+- **`fakeSourceControlFactory`** — seeds each queue with a deterministic single-commit history so ingest resolves a stable head URI (and re-ingesting the same queue exercises the dedup path). A real deployment supplies a VCS-backed `sourcecontrol.Factory`, which is also where a queue's promotion ref is resolved. The fake has no ref to move, so a promotion locally shows up only in the record consumer's logs.
 
 ## Layout
 

service/stovepipe/server/main.go

@@ -130,7 +130,10 @@ func (f *inMemoryCounterFactory) For(config counter.Config) (counter.Counter, er
 
 // fakeSourceControlFactory is the example SourceControl factory. It seeds each queue with a
 // deterministic single-commit history so ingest resolves a stable head URI (and re-ingesting
-// the same queue exercises the dedup path). A real deployment supplies a VCS-backed factory.
+// the same queue exercises the dedup path). It has no ref to promote onto, so a promotion
+// only succeeds or reports the commit as gone, and the local stack shows it in the record
+// consumer's log. A real deployment supplies a VCS-backed factory, which is also where the
+// promotion ref is resolved from the queue name, alongside the repo and credentials.
 type fakeSourceControlFactory struct{}
 
 func (fakeSourceControlFactory) For(cfg sourcecontrol.Config) (sourcecontrol.SourceControl, error) {
@@ -426,7 +429,7 @@ func registerPrimaryControllers(
 	}
 	count++
 
-	recordController := record.NewController(logger, scope, store, stovepipemq.TopicKeyRecord, "stovepipe-record")
+	recordController := record.NewController(logger, scope, store, scf, stovepipemq.TopicKeyRecord, "stovepipe-record")
 	if err := c.Register(recordController); err != nil {
 		return count, fmt.Errorf("failed to register record controller: %w", err)
 	}

stovepipe/controller/record/BUILD.bazel

@@ -11,6 +11,7 @@ go_library(
         "//stovepipe/core/loader:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity: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",
@@ -26,6 +27,8 @@ go_test(
         "//platform/consumer/mock:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity: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/record/record.go

@@ -18,7 +18,9 @@
 //
 // The durable state is a ValidationFact per validated commit, plus the queue's
 // last-green bookmark, which process reads to choose an incremental build
-// baseline. Downstream hooks are not implemented yet.
+// baseline. A green commit is also promoted, moving the queue's promotion ref so
+// downstream systems can pull the latest green commit by name. Downstream hooks
+// are not implemented yet.
 package record
 
 import (
@@ -33,17 +35,19 @@ import (
 	"github.com/uber/submitqueue/stovepipe/core/loader"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 	"go.uber.org/zap"
 )
 
 // Controller consumes Record messages, records the build's validation fact, and
-// advances the queue's last-green bookmark when that fact is green. Implements
-// consumer.Controller.
+// when that fact is green advances the queue's last-green bookmark and promotes
+// the commit. Implements consumer.Controller.
 type Controller struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
 	stores        storage.Factory
+	sourceControl sourcecontrol.Factory
 	topicKey      consumer.TopicKey
 	consumerGroup string
 }
@@ -64,21 +68,24 @@ func NewController(
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
 	stores storage.Factory,
+	sourceControl sourcecontrol.Factory,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
 ) *Controller {
 	return &Controller{
 		logger:        logger.Named("record_controller"),
 		metricsScope:  scope.SubScope("record_controller"),
 		stores:        stores,
+		sourceControl: sourceControl,
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
 	}
 }
 
-// Process loads the request referenced by the delivery and, when its build
-// succeeded, advances the queue's last-green bookmark. Returns nil to ack
-// (success) or an error to nack (retry) / reject (DLQ).
+// Process loads the request referenced by the delivery, records its validation
+// fact and, when that fact is green, advances the queue's last-green bookmark and
+// promotes the commit. Returns nil to ack (success) or an error to nack (retry) /
+// reject (DLQ).
 //
 // buildsignal stamps the outcome on the request before publishing here, so a
 // request without a build outcome is a producer invariant violation rather
@@ -123,11 +130,18 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 			metrics.NamedCounter(c.metricsScope, _opName, "not_green", 1)
 			return nil
 		}
-		if err := c.advanceLastGreen(ctx, store, request); err != nil {
+		holdsBookmark, err := c.advanceLastGreen(ctx, store, request)
+		if err != nil {
 			metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1)
 			return err
 		}
-		return nil
+		if !holdsBookmark {
+			// A later green commit already holds the bookmark, so it also owns
+			// the promotion ref: promoting this older commit would move the ref
+			// backwards.
+			return nil
+		}
+		return c.promote(ctx, request)
 
 	case entity.RequestStateCancelled:
 		// A cancelled build decided nothing about the commit, so it establishes
@@ -213,30 +227,34 @@ func degreeFor(state entity.RequestState) float64 {
 }
 
 // advanceLastGreen points the queue's bookmark at request, retrying on version
-// conflicts. The bookmark only moves forward: a candidate whose id is not newer
-// than the stored one is skipped without a write, which also makes a redelivery
-// of the same request a no-op.
+// conflicts, and reports whether request holds the bookmark afterwards. The
+// bookmark only moves forward: an older candidate is skipped without a write and
+// does not hold it, while a redelivery of the request that already set it holds it
+// without a write, so the promotion that follows is retried.
 //
 // The bookmark is a cache of "newest green URI" derived from the facts, so it is
 // advanced only after the green fact is durable. Losing the advance to a crash is
 // recoverable — the redelivery reloads the same fact and retries — whereas a
 // bookmark with no fact behind it would point at greenness nothing recorded.
-func (c *Controller) advanceLastGreen(ctx context.Context, store storage.Storage, request entity.Request) error {
+func (c *Controller) advanceLastGreen(ctx context.Context, store storage.Storage, request entity.Request) (bool, error) {
 	queueStore := store.GetQueueStore()
 
 	for {
 		queueRow, err := queueStore.Get(ctx, request.Queue)
 		if err != nil {
-			return fmt.Errorf("failed to load queue %s to advance last green: %w", request.Queue, err)
+			return false, fmt.Errorf("failed to load queue %s to advance last green: %w", request.Queue, err)
 		}
 
-		newer, err := isNewerRequest(request.Queue, request.ID, queueRow.LastGreenRequestID)
+		cmp, err := compareToBookmark(request.Queue, request.ID, queueRow.LastGreenRequestID)
 		if err != nil {
 			// Non-retryable: re-parsing the same ids cannot start succeeding.
-			return err
+			return false, err
 		}
-		if !newer {
-			return nil
+		if cmp < 0 {
+			return false, nil
+		}
+		if cmp == 0 {
+			return true, nil
 		}
 
 		updated := queueRow
@@ -247,7 +265,7 @@ func (c *Controller) advanceLastGreen(ctx context.Context, store storage.Storage
 			if errors.Is(err, storage.ErrVersionMismatch) {
 				continue
 			}
-			return fmt.Errorf("failed to advance last green for queue %s: %w", request.Queue, err)
+			return false, fmt.Errorf("failed to advance last green for queue %s: %w", request.Queue, err)
 		}
 
 		metrics.NamedCounter(c.metricsScope, _opName, "last_green_advanced", 1)
@@ -256,21 +274,68 @@ func (c *Controller) advanceLastGreen(ctx context.Context, store storage.Storage
 			"request_id", request.ID,
 			"last_green_uri", request.URI,
 		)
-		return nil
+		return true, nil
 	}
 }
 
-// isNewerRequest reports whether candidate was ingested after current. An empty
-// current means the bookmark has never been set, so any candidate is newer.
-func isNewerRequest(queue, candidate, current string) (bool, error) {
+// promote points the queue's promotion ref at the request's commit so downstream
+// systems can pull the latest green commit by name. Which ref that is — and whether
+// the queue has one at all — is source-control configuration, so this stage names
+// only the commit.
+//
+// Like the bookmark, the ref is a cache of the facts, so it moves only after the
+// green fact is durable. Promotion is idempotent, so a redelivery repeats it
+// harmlessly. A commit that a rewritten history dropped from the ref cannot be
+// promoted by any retry, so that case is counted and skipped rather than failed.
+func (c *Controller) promote(ctx context.Context, request entity.Request) error {
+	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("failed to resolve source control for queue %s: %w", request.Queue, err)
+	}
+
+	if err := sc.Promote(ctx, request.URI); err != nil {
+		if sourcecontrol.IsNotFound(err) {
+			metrics.NamedCounter(c.metricsScope, _opName, "promotions_skipped", 1,
+				metrics.NewTag("reason", "unknown_uri"),
+			)
+			c.logger.Warnw("green commit is no longer on the queue's ref; skipping promotion",
+				"queue", request.Queue,
+				"request_id", request.ID,
+				"uri", request.URI,
+			)
+			return nil
+		}
+
+		metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1,
+			metrics.NewTag("stage", "promote"),
+		)
+		return fmt.Errorf("failed to promote uri %s of queue %s: %w", request.URI, request.Queue, err)
+	}
+
+	metrics.NamedCounter(c.metricsScope, _opName, "promotions", 1)
+	c.logger.Infow("promoted green commit",
+		"queue", request.Queue,
+		"request_id", request.ID,
+		"uri", request.URI,
+	)
+	return nil
+}
+
+// compareToBookmark orders candidate against the request id currently holding the
+// bookmark, by ingest order, using the sign convention of entity.CompareRequestID.
+// An empty current means the bookmark has never been set, so any candidate is newer.
+func compareToBookmark(queue, candidate, current string) (int, error) {
 	if current == "" {
-		return true, nil
+		return 1, nil
 	}
 	cmp, err := entity.CompareRequestID(queue, candidate, current)
 	if err != nil {
-		return false, fmt.Errorf("failed to compare request ids for queue %s: %w", queue, err)
+		return 0, fmt.Errorf("failed to compare request ids for queue %s: %w", queue, err)
 	}
-	return cmp > 0, nil
+	return cmp, nil
 }
 
 // loadRequest loads the request by id.