feat(stovepipe): apply queue transitions and materialize history

Makefile

@@ -575,9 +575,10 @@ local-stop: ## Stop every local stack — SubmitQueue, Stovepipe, and Runway
 
 local-stovepipe-debug-start: build-stovepipe-linux-debug ## Start Stovepipe under delve in Docker (attach IDE to :2345)
 	@echo "Starting Stovepipe service with compose (debug)..."
-	@$(COMPOSE) -f $(STOVEPIPE_COMPOSE_FILE) -f $(STOVEPIPE_DEBUG_COMPOSE_FILE) -p $(STOVEPIPE_LOCAL_PROJECT) up -d --build --wait
+	@$(COMPOSE) -f $(STOVEPIPE_COMPOSE_FILE) -f $(STOVEPIPE_DEBUG_COMPOSE_FILE) -p $(STOVEPIPE_LOCAL_PROJECT) up -d --build --wait mysql-app mysql-queue
 	@echo "Applying storage and queue schemas..."
 	@$(MAKE) -s local-init-stovepipe-schemas
+	@$(COMPOSE) -f $(STOVEPIPE_COMPOSE_FILE) -f $(STOVEPIPE_DEBUG_COMPOSE_FILE) -p $(STOVEPIPE_LOCAL_PROJECT) up -d --build --wait
 	@echo ""
 	@echo "✅ Stovepipe debug is running (delve :2345)!"
 	@echo ""
@@ -591,9 +592,10 @@ local-stovepipe-logs: ## View logs from the running Stovepipe 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
+	@$(COMPOSE) -f $(STOVEPIPE_COMPOSE_FILE) -p $(STOVEPIPE_LOCAL_PROJECT) up -d --build --wait mysql-app mysql-queue
 	@echo "Applying storage and queue schemas..."
 	@$(MAKE) -s local-init-stovepipe-schemas
+	@$(COMPOSE) -f $(STOVEPIPE_COMPOSE_FILE) -p $(STOVEPIPE_LOCAL_PROJECT) up -d --build --wait
 	@echo ""
 	@echo "✅ Stovepipe service is running!"
 	@echo ""

service/stovepipe/docker-compose.yml

@@ -14,7 +14,7 @@
 # Quick start:
 #   make local-stovepipe-start
 #
-# After `up`, the storage and queue schemas are applied (local-init-stovepipe-schemas).
+# Start the databases, apply schemas (local-init-stovepipe-schemas), then start the service.
 
 services:
   # Storage database - Stovepipe's request and request_uri tables.

service/stovepipe/server/BUILD.bazel

@@ -31,6 +31,8 @@ go_library(
         "//stovepipe/controller/process:go_default_library",
         "//stovepipe/controller/record:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
+        "//stovepipe/core/queuelog:go_default_library",
+        "//stovepipe/core/queuestate:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/extension/buildrunner:go_default_library",
         "//stovepipe/extension/buildrunner/fake:go_default_library",

service/stovepipe/server/main.go

@@ -53,6 +53,8 @@ import (
 	"github.com/uber/submitqueue/stovepipe/controller/process"
 	"github.com/uber/submitqueue/stovepipe/controller/record"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
+	"github.com/uber/submitqueue/stovepipe/core/queuelog"
+	"github.com/uber/submitqueue/stovepipe/core/queuestate"
 	"github.com/uber/submitqueue/stovepipe/core/requestlog"
 	"github.com/uber/submitqueue/stovepipe/extension/buildrunner"
 	buildrunnerfake "github.com/uber/submitqueue/stovepipe/extension/buildrunner/fake"
@@ -245,6 +247,12 @@ func run() error {
 	brf := fakeBuildRunnerFactory{}
 
 	storageFty := storageFactory{backend: store}
+	stateTransitions := queuestate.NewTransitionController(storageFty, sourceControl, queuelog.NewMaterializer(scope))
+	for _, queue := range tenants {
+		if _, err := stateTransitions.Initialize(ctx, queue); err != nil {
+			return fmt.Errorf("initialize queue enablement %q: %w", queue, err)
+		}
+	}
 	materializer := requestlog.NewMaterializer(scope)
 	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, materializer, registry, sourceControl, brf, hookResolver{})
 	if err != nil {

stovepipe/core/queuelog/BUILD.bazel

@@ -0,0 +1,35 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "materializer.go",
+        "queue_summary.go",
+    ],
+    importpath = "github.com/uber/submitqueue/stovepipe/core/queuelog",
+    visibility = [
+        "//service/stovepipe:__subpackages__",
+        "//stovepipe:__subpackages__",
+        "//test:__subpackages__",
+    ],
+    deps = [
+        "//platform/metrics:go_default_library",
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["materializer_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "//stovepipe/extension/storage/mock:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+    ],
+)

stovepipe/core/queuelog/materializer.go

@@ -0,0 +1,125 @@
+// Copyright (c) 2026 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package queuelog retains queue history and projects its current status summary.
+package queuelog
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math"
+
+	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/platform/metrics"
+	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/storage"
+)
+
+// ErrInconsistentHistory identifies missing or conflicting queue history or summaries.
+var ErrInconsistentHistory = errors.New("inconsistent queue history")
+
+// Materializer retains committed state changes and updates their summary.
+type Materializer interface {
+	// PersistLog idempotently retains an occurrence, then projects its state.
+	PersistLog(context.Context, storage.Storage, entity.QueueLog) error
+}
+
+type materializer struct{ scope tally.Scope }
+
+func NewMaterializer(scope tally.Scope) Materializer {
+	return &materializer{scope: scope.SubScope("queue_log_materializer")}
+}
+
+func NewQueueStateLog(queue entity.Queue) entity.QueueLog {
+	return entity.QueueLog{ID: entity.QueueLogID(queue.EnablementVersion), Queue: queue.Name,
+		State: queue.Enablement, QueueStateVersion: queue.EnablementVersion,
+		EffectiveFromCommitURI: queue.EffectiveFromCommitURI, TimestampMs: queue.EnablementChangedAtMs}
+}
+
+func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, log entity.QueueLog) (retErr error) {
+	op := metrics.Begin(m.scope, "persist_log", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...)
+	defer func() { op.Complete(retErr) }()
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	if err := log.Validate(); err != nil {
+		return fmt.Errorf("queue %q: invalid log %q at state version %d: %w: %v", log.Queue, log.ID, log.QueueStateVersion, ErrInconsistentHistory, err)
+	}
+	logs := stores.GetQueueLogStore()
+	if err := logs.Create(ctx, log); err != nil {
+		if !errors.Is(err, storage.ErrAlreadyExists) {
+			return fmt.Errorf("retain queue log: %w", err)
+		}
+		retained, err := logs.Get(ctx, log.ID)
+		if err != nil {
+			return fmt.Errorf("reconcile queue log: %w", err)
+		}
+		if retained != log {
+			return fmt.Errorf("queue %q: log %q at state version %d conflicts with retained log %q for queue %q at state version %d: %w", log.Queue, log.ID, log.QueueStateVersion, retained.ID, retained.Queue, retained.QueueStateVersion, ErrInconsistentHistory)
+		}
+		log = retained
+	}
+	return materializeQueueSummary(ctx, stores.GetQueueSummaryStore(), log)
+}
+
+func materializeQueueSummary(ctx context.Context, store storage.QueueSummaryStore, log entity.QueueLog) error {
+	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		current, err := store.Get(ctx)
+		if storage.IsNotFound(err) {
+			initial := entity.QueueSummary{Queue: log.Queue, State: log.State,
+				QueueStateVersion: log.QueueStateVersion, EffectiveFromCommitURI: log.EffectiveFromCommitURI,
+				StateTimestampMs: log.TimestampMs, Version: 1}
+			if err := store.Create(ctx, initial); errors.Is(err, storage.ErrAlreadyExists) {
+				continue
+			} else {
+				return err
+			}
+		}
+		if err != nil {
+			return err
+		}
+		if current.Queue != log.Queue {
+			return fmt.Errorf("queue %q log %q: summary belongs to queue %q at projection version %d: %w", log.Queue, log.ID, current.Queue, current.Version, ErrInconsistentHistory)
+		}
+		if err := current.Validate(); err != nil {
+			return fmt.Errorf("queue %q: invalid summary at projection version %d, state version %d: %w: %v", log.Queue, current.Version, current.QueueStateVersion, ErrInconsistentHistory, err)
+		}
+		if log.QueueStateVersion < current.QueueStateVersion {
+			return nil
+		}
+		if log.QueueStateVersion == current.QueueStateVersion {
+			if current.Snapshot() != log.Snapshot() {
+				return fmt.Errorf("queue %q: log %q conflicts with summary state version %d (projection version %d): %w", log.Queue, log.ID, current.QueueStateVersion, current.Version, ErrInconsistentHistory)
+			}
+			return nil
+		}
+		if current.Version == math.MaxInt64 {
+			return fmt.Errorf("queue %q log %q: summary projection version %d exhausted: %w", log.Queue, log.ID, current.Version, ErrInconsistentHistory)
+		}
+		updated := current
+		updated.State, updated.QueueStateVersion = log.State, log.QueueStateVersion
+		updated.EffectiveFromCommitURI, updated.StateTimestampMs = log.EffectiveFromCommitURI, log.TimestampMs
+		newVersion := current.Version + 1
+		if err := store.Update(ctx, updated, current.Version, newVersion); errors.Is(err, storage.ErrVersionMismatch) {
+			continue
+		} else if err != nil {
+			return err
+		}
+		return nil
+	}
+}