feat(stovepipe): wire SourceControl + stores into ingest, publish to process

## Summary

### Why?

Ingest was a thin stub: it minted a request id but never resolved the commit URI, persisted anything, or moved the request onto the pipeline, so `Request.URI` stayed empty and nothing consumed the work. This makes ingest the real pipeline entry and adds the first internal queue contract so the pipeline can hand work to the next stage.

### What?

Ingest now resolves the queue's head URI via the SourceControl extension, dedups on the (queue, URI) pair, persists the Request and its URI mapping via storage, and publishes the request id to a new process stage over the messaging queue. Ingestion is idempotent: a re-reported head resolves to the already-minted request and nothing is published again. The URI mapping is claimed before the request row is written, so a lost race leaves no orphan row.

Adds the first internal proto message-queue contract under `stovepipe/core/messagequeue` (proto3 + protojson, mirroring `api/runway/messagequeue`): a `ProcessRequest` payload carrying the id, the `TopicKeyProcess` constant, and the protojson glue, wired into the proto codegen (`tool/proto`, `PROTO_PACKAGES`). Per CLAUDE.md, internal contracts live under the domain's `core/`, not `api/`, and the contract package owns both the payload and its topic keys.

Adds a minimal `process` consumer (`stovepipe/controller/process`) that reloads the Request by id and logs it; a not-yet-visible request is retryable so redelivery converges. The build-strategy/ancestry logic the RFC assigns to `process` is deferred.

Wires the example server (`example/stovepipe/server`) into a MySQL storage + MySQL queue + fake SourceControl stack with the process consumer running, plus docker-compose (two databases) and a schema-init make target.

## Test Plan

✅ `bazel test //stovepipe/...` — contract round-trip + topic-key binding, ingest (happy/dedup/race/unknown-queue/infra-error paths), and process consumer unit tests.
✅ `bazel test //test/integration/stovepipe:stovepipe_test` — compose-backed: Ingest persists the request + URI mapping, publishes to the process topic, and a re-ingest dedups to the same id.
✅ `bazel build //...`, `make fmt`.

Author: behinddwalls

Makefile

@@ -35,7 +35,7 @@ GOIMPORTS_VERSION ?= v0.33.0
 # (the out_dir convention in tool/proto/BUILD.bazel) and copied back here. A
 # package may hold multiple .proto files (e.g. an RPC contract plus messagequeue
 # contracts); all generated stubs land in the same protopb/ dir.
-PROTO_PACKAGES = api/base/change api/base/mergestrategy api/base/messagequeue api/runway/messagequeue api/runway api/submitqueue/gateway api/submitqueue/orchestrator api/stovepipe
+PROTO_PACKAGES = api/base/change api/base/mergestrategy api/base/messagequeue api/runway/messagequeue api/runway api/submitqueue/gateway api/submitqueue/orchestrator api/stovepipe stovepipe/core/messagequeue
 
 # Set REPO_ROOT for docker-compose
 export REPO_ROOT := $(shell pwd)
@@ -51,7 +51,7 @@ define assert_clean
 	fi
 endef
 
-.PHONY: build build-all-linux build-runway-linux build-submitqueue-gateway-linux build-submitqueue-orchestrator-linux build-stovepipe-linux check-gazelle check-mocks check-tidy clean clean-proto deps e2e-test fmt gazelle integration-test integration-test-submitqueue-consumer integration-test-extensions integration-test-submitqueue-gateway integration-test-submitqueue-orchestrator license-fix lint lint-fmt lint-license local-init-runway-queue-schema local-runway-start local-runway-stop local-submitqueue-clean local-submitqueue-gateway-start local-submitqueue-gateway-stop local-init-submitqueue-schemas local-submitqueue-logs local-submitqueue-orchestrator-start local-submitqueue-orchestrator-stop local-submitqueue-ps local-submitqueue-restart local-submitqueue-start local-stop local-stovepipe-logs local-stovepipe-start local-stovepipe-stop mocks proto query-deps query-targets run-client-runway run-client-submitqueue-gateway run-client-submitqueue-orchestrator run-client-stovepipe run-queue-admin test test-no-cache tidy tidy-bazel tidy-go help
+.PHONY: build build-all-linux build-runway-linux build-submitqueue-gateway-linux build-submitqueue-orchestrator-linux build-stovepipe-linux check-gazelle check-mocks check-tidy clean clean-proto deps e2e-test fmt gazelle integration-test integration-test-submitqueue-consumer integration-test-extensions integration-test-submitqueue-gateway integration-test-submitqueue-orchestrator license-fix lint lint-fmt lint-license local-init-runway-queue-schema local-init-stovepipe-schemas local-runway-start local-runway-stop local-submitqueue-clean local-submitqueue-gateway-start local-submitqueue-gateway-stop local-init-submitqueue-schemas local-submitqueue-logs local-submitqueue-orchestrator-start local-submitqueue-orchestrator-stop local-submitqueue-ps local-submitqueue-restart local-submitqueue-start local-stop local-stovepipe-logs local-stovepipe-start local-stovepipe-stop mocks proto query-deps query-targets run-client-runway run-client-submitqueue-gateway run-client-submitqueue-orchestrator run-client-stovepipe run-queue-admin test test-no-cache tidy tidy-bazel tidy-go help
 
 
 build: ## Build all services and examples
@@ -221,6 +221,19 @@ local-init-runway-queue-schema: ## Apply queue schema only (mysql-queue) for Run
 	done
 	@echo "✅ Runway queue schema applied successfully"
 
+local-init-stovepipe-schemas: ## Apply storage (mysql-app) and queue (mysql-queue) schemas for Stovepipe compose stacks
+	@echo "Applying storage schema to mysql-app..."
+	@for file in stovepipe/extension/storage/mysql/schema/*.sql; do \
+		echo "  - Applying $$(basename $$file)..."; \
+		docker exec -i $(STOVEPIPE_LOCAL_PROJECT)-mysql-app-1 mysql -uroot -proot submitqueue < $$file 2>&1 | grep -v "Using a password" || true; \
+	done
+	@echo "Applying queue schema to mysql-queue..."
+	@for file in platform/extension/messagequeue/mysql/schema/*.sql; do \
+		echo "  - Applying $$(basename $$file)..."; \
+		docker exec -i $(STOVEPIPE_LOCAL_PROJECT)-mysql-queue-1 mysql -uroot -proot submitqueue < $$file 2>&1 | grep -v "Using a password" || true; \
+	done
+	@echo "✅ Stovepipe schemas applied successfully"
+
 local-runway-start: build-runway-linux ## Start Runway locally (runway + MySQL queue)
 	@echo "Starting Runway with compose..."
 	@$(COMPOSE) -f $(RUNWAY_COMPOSE_FILE) -p $(RUNWAY_LOCAL_PROJECT) up -d --build --wait
@@ -310,9 +323,11 @@ local-stop: ## Stop all services (keep data)
 local-stovepipe-logs: ## View logs from the running Stovepipe service
 	@$(COMPOSE) -f $(STOVEPIPE_COMPOSE_FILE) -p $(STOVEPIPE_LOCAL_PROJECT) logs -f
 
-local-stovepipe-start: build-stovepipe-linux ## Start Stovepipe service (single Ping-only gRPC service)
+local-stovepipe-start: build-stovepipe-linux ## Start Stovepipe service (gRPC service + MySQL storage + MySQL queue)
 	@echo "Starting Stovepipe service with compose..."
 	@$(COMPOSE) -f $(STOVEPIPE_COMPOSE_FILE) -p $(STOVEPIPE_LOCAL_PROJECT) up -d --build --wait
+	@echo "Applying storage and queue schemas..."
+	@$(MAKE) -s local-init-stovepipe-schemas
 	@echo ""
 	@echo "✅ Stovepipe service is running!"
 	@echo ""

example/stovepipe/docker-compose.yml

@@ -1,7 +1,9 @@
 # Docker Compose for the Stovepipe service.
 #
-# Stovepipe is currently a single Ping-only gRPC service with no storage or
-# queue dependencies, so this stack runs just the one service.
+# Stovepipe ingests a queue's head commit, persists the Request, and publishes it
+# to the process stage over a messaging queue; the process consumer reloads the
+# Request. The stack therefore runs the service plus two MySQL databases: one for
+# storage (request, request_uri) and one for the messaging queue.
 #
 # IMPORTANT: Before running compose, build the Linux binary:
 #   make build-stovepipe-linux
@@ -12,8 +14,40 @@
 # Quick start:
 #   make local-stovepipe-start
 #
+# After `up`, the storage and queue schemas are applied (local-init-stovepipe-schemas).
 
 services:
+  # Storage database - Stovepipe's request and request_uri tables.
+  mysql-app:
+    image: mysql:8.0
+    environment:
+      MYSQL_ROOT_PASSWORD: root
+      MYSQL_DATABASE: submitqueue
+    ports:
+      - "3306" # Random ephemeral port to avoid conflicts
+    healthcheck:
+      # Use 127.0.0.1 (TCP) instead of localhost (Unix socket). MySQL treats
+      # "localhost" as a socket connection, which can be ready before the TCP
+      # listener — causing dependent services that connect over TCP to fail.
+      test: ["CMD", "mysqladmin", "ping", "-h", "127.0.0.1", "-proot"]
+      interval: 5s
+      timeout: 5s
+      retries: 10
+
+  # Queue database - messaging infrastructure (messages, offsets, partition leases).
+  mysql-queue:
+    image: mysql:8.0
+    environment:
+      MYSQL_ROOT_PASSWORD: root
+      MYSQL_DATABASE: submitqueue
+    ports:
+      - "3306" # Random ephemeral port to avoid conflicts
+    healthcheck:
+      test: ["CMD", "mysqladmin", "ping", "-h", "127.0.0.1", "-proot"]
+      interval: 5s
+      timeout: 5s
+      retries: 10
+
   stovepipe-service:
     build:
       context: ${REPO_ROOT}
@@ -22,3 +56,11 @@ services:
       - "8080" # Random ephemeral port to avoid conflicts
     environment:
       - PORT=:8080
+      - STORAGE_MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true
+      - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true
+      - HOSTNAME=stovepipe-dev
+    depends_on:
+      mysql-app:
+        condition: service_healthy
+      mysql-queue:
+        condition: service_healthy

example/stovepipe/server/BUILD.bazel

@@ -7,7 +7,19 @@ go_library(
     visibility = ["//visibility:private"],
     deps = [
         "//api/stovepipe/protopb",
+        "//platform/consumer",
+        "//platform/errs",
+        "//platform/errs/generic",
+        "//platform/errs/mysql",
+        "//platform/extension/messagequeue",
+        "//platform/extension/messagequeue/mysql",
         "//stovepipe/controller",
+        "//stovepipe/controller/process",
+        "//stovepipe/core/messagequeue",
+        "//stovepipe/extension/sourcecontrol",
+        "//stovepipe/extension/sourcecontrol/fake",
+        "//stovepipe/extension/storage/mysql",
+        "@com_github_go_sql_driver_mysql//:mysql",
         "@com_github_uber_go_tally//:tally",
         "@org_golang_google_grpc//:grpc",
         "@org_golang_google_grpc//reflection",

example/stovepipe/server/main.go

@@ -16,6 +16,7 @@ package main
 
 import (
 	"context"
+	"database/sql"
 	"errors"
 	"fmt"
 	"net"
@@ -25,9 +26,21 @@ import (
 	"syscall"
 	"time"
 
+	_ "github.com/go-sql-driver/mysql"
 	"github.com/uber-go/tally"
 	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
+	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/platform/errs"
+	genericerrs "github.com/uber/submitqueue/platform/errs/generic"
+	mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql"
+	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
+	queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
 	"github.com/uber/submitqueue/stovepipe/controller"
+	"github.com/uber/submitqueue/stovepipe/controller/process"
+	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
+	"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
+	sourcecontrolfake "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/fake"
+	storageMySQL "github.com/uber/submitqueue/stovepipe/extension/storage/mysql"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/reflection"
@@ -70,6 +83,15 @@ func (c *inMemoryCounter) Next(_ context.Context, domain string) (int64, error)
 	return c.values[domain], nil
 }
 
+// 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.
+type fakeSourceControlFactory struct{}
+
+func (fakeSourceControlFactory) For(cfg sourcecontrol.Config) (sourcecontrol.SourceControl, error) {
+	return sourcecontrolfake.New([]string{fmt.Sprintf("git://%s/HEAD", cfg.QueueName)}), nil
+}
+
 func main() {
 	code := 0
 	if err := run(); err != nil {
@@ -131,12 +153,85 @@ func run() error {
 		metricsWgDone.Wait()
 	}()
 
+	// Storage database (request + request_uri tables).
+	storageDSN := os.Getenv("STORAGE_MYSQL_DSN")
+	if storageDSN == "" {
+		return fmt.Errorf("STORAGE_MYSQL_DSN environment variable is required")
+	}
+	storageDB, err := sql.Open("mysql", storageDSN)
+	if err != nil {
+		return fmt.Errorf("failed to open storage database: %w", err)
+	}
+	defer storageDB.Close()
+
+	store, err := storageMySQL.NewStorage(storageDB, scope.SubScope("storage"))
+	if err != nil {
+		return fmt.Errorf("failed to create storage: %w", err)
+	}
+	defer store.Close()
+
+	// Queue database (messaging infrastructure for the process stage).
+	queueDSN := os.Getenv("QUEUE_MYSQL_DSN")
+	if queueDSN == "" {
+		return fmt.Errorf("QUEUE_MYSQL_DSN environment variable is required")
+	}
+	queueDB, err := sql.Open("mysql", queueDSN)
+	if err != nil {
+		return fmt.Errorf("failed to open queue database: %w", err)
+	}
+	defer queueDB.Close()
+
+	mysqlQueue, err := queueMySQL.NewQueue(queueMySQL.Params{
+		DB:           queueDB,
+		Logger:       logger,
+		MetricsScope: scope.SubScope("queue"),
+	})
+	if err != nil {
+		return fmt.Errorf("failed to create queue: %w", err)
+	}
+	defer mysqlQueue.Close()
+
+	subscriberName := os.Getenv("HOSTNAME")
+	if subscriberName == "" {
+		subscriberName = fmt.Sprintf("stovepipe-%d", time.Now().Unix())
+	}
+
+	registry, err := newTopicRegistry(mysqlQueue, subscriberName)
+	if err != nil {
+		return fmt.Errorf("failed to create topic registry: %w", err)
+	}
+
+	// Consumer running the process stage.
+	primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry,
+		errs.NewClassifierProcessor(
+			genericerrs.Classifier,
+			mysqlerrs.Classifier,
+		),
+	)
+
+	processController := process.NewController(logger.Sugar(), scope, store, stovepipemq.TopicKeyProcess, "stovepipe-process")
+	if err := primaryConsumer.Register(processController); err != nil {
+		return fmt.Errorf("failed to register process controller: %w", err)
+	}
+
+	if err := primaryConsumer.Start(ctx); err != nil {
+		return fmt.Errorf("failed to start consumer: %w", err)
+	}
+	logger.Info("consumer started")
+
 	// Create gRPC server
 	grpcServer := grpc.NewServer()
 
 	// Create controllers and wrap them for gRPC
 	pingController := controller.NewPingController(logger, scope)
-	ingestController := controller.NewIngestController(logger.Sugar(), scope, newInMemoryCounter())
+	ingestController := controller.NewIngestController(
+		logger.Sugar(),
+		scope,
+		newInMemoryCounter(),
+		fakeSourceControlFactory{},
+		store,
+		registry,
+	)
 	srv := &StovepipeServer{
 		pingController:   pingController,
 		ingestController: ingestController,
@@ -166,28 +261,50 @@ func run() error {
 	}()
 
 	// Wait for interrupt signal or server critical error
-	// If interruption is signaled, gracefully stop the server
-	// If an error happens during shutdown, return the actual error, not the context cancellation error
 	var serverErr error
 	select {
 	case <-ctx.Done():
 		fmt.Println("Shutting down stovepipe server due to interruption signal...")
 
 		// Set the error to the context cancellation error to be surfaced as a desired exit code by the main function
-		// to indicate that the server was stopped as intended
-		// It may be overridden by the server error if any
+		// to indicate that the server was stopped as intended. It may be overridden by the server error if any.
 		err = ctx.Err()
 
 		// stop GRPC server and wait for it to exit
 		grpcServer.GracefulStop()
 		serverErr = <-serverErrCh
 	case serverErr = <-serverErrCh:
 		fmt.Println("Shutting down stovepipe server due to critical GRPC server error...")
+		cancel()
 	}
 
 	if serverErr != nil {
-		err = fmt.Errorf("GRPC server exited with error: %w", serverErr)
+		serverErr = fmt.Errorf("GRPC server exited with error: %w", serverErr)
+	}
+
+	consumerStopErr := primaryConsumer.Stop(30000)
+	if consumerStopErr != nil {
+		consumerStopErr = fmt.Errorf("failed to stop consumer: %w", consumerStopErr)
+	}
+
+	if consumerStopErr != nil || serverErr != nil {
+		err = errors.Join(err, consumerStopErr, serverErr)
 	}
 
 	return err
 }
+
+// newTopicRegistry builds the TopicRegistry for Stovepipe's internal pipeline queues. ingest
+// publishes to the process topic and the process consumer subscribes to it.
+func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) {
+	return consumer.NewTopicRegistry([]consumer.TopicConfig{
+		{
+			Key:   stovepipemq.TopicKeyProcess,
+			Name:  "process",
+			Queue: q,
+			Subscription: extqueue.DefaultSubscriptionConfig(
+				subscriberName, "stovepipe-process",
+			),
+		},
+	})
+}

stovepipe/controller/BUILD.bazel

@@ -10,10 +10,15 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//api/stovepipe/protopb",
+        "//platform/base/messagequeue",
+        "//platform/consumer",
         "//platform/errs",
         "//platform/extension/counter",
         "//platform/metrics",
+        "//stovepipe/core/messagequeue",
         "//stovepipe/entity",
+        "//stovepipe/extension/sourcecontrol",
+        "//stovepipe/extension/storage",
         "@com_github_uber_go_tally//:tally",
         "@org_uber_go_zap//:zap",
     ],
@@ -28,7 +33,14 @@ go_test(
     embed = [":controller"],
     deps = [
         "//api/stovepipe/protopb",
+        "//platform/consumer",
         "//platform/extension/counter/mock",
+        "//platform/extension/messagequeue/mock",
+        "//stovepipe/core/messagequeue",
+        "//stovepipe/extension/sourcecontrol",
+        "//stovepipe/extension/sourcecontrol/mock",
+        "//stovepipe/extension/storage",
+        "//stovepipe/extension/storage/mock",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
         "@com_github_uber_go_tally//:tally",