feat(stovepipe): look up current and commit-specific queue status

service/stovepipe/README.md

@@ -105,3 +105,7 @@ grpcurl -plaintext -d '{"message": "hello"}' localhost:8083 uber.submitqueue.sto
 ## Shutdown
 
 The server handles `SIGINT` / `SIGTERM` gracefully: it drains in-flight RPCs, then stops the primary pipeline consumer followed by the DLQ consumer (30-second limit for each). It exits `0` on clean shutdown, `143` (128 + SIGTERM) when stopped by signal, and `1` on startup/runtime errors (details on stderr). Shutdown errors override the signal exit code.
+
+## Queue status
+
+Queue enablement is stored separately from validation results. Startup registers configured queues as disabled and preserves any registered state. `GetQueueStatus` returns the current summary, or the state covering a supplied commit through the existing source-control integration. History and summary writes can be retried after an interruption, so the summary may briefly lag internal state. Apply the schemas before starting the service; the local Make targets handle this order. See [queue status](../../doc/rfc/stovepipe/queue-state.md).

service/stovepipe/server/handler/BUILD.bazel

@@ -9,6 +9,9 @@ go_library(
         "//api/stovepipe/protopb:go_default_library",
         "//service/stovepipe/server/mapper:go_default_library",
         "//stovepipe/controller:go_default_library",
+        "//stovepipe/core/queuelog:go_default_library",
+        "@org_golang_google_grpc//codes:go_default_library",
+        "@org_golang_google_grpc//status:go_default_library",
     ],
 )
 
@@ -22,8 +25,10 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//api/stovepipe/protopb:go_default_library",
+        "//platform/errs:go_default_library",
         "//stovepipe/controller:go_default_library",
         "//stovepipe/controller/mock:go_default_library",
+        "//stovepipe/core/queuelog:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",

service/stovepipe/server/handler/list_test.go

@@ -68,7 +68,7 @@ func TestList(t *testing.T) {
 				}, req)
 				return tt.result, tt.err
 			}}
-			srv := NewStovepipeServer(nil, nil, nil, nil, fake)
+			srv := NewStovepipeServer(nil, nil, nil, nil, fake, nil)
 			response, err := srv.List(ctx, &pb.ListRequest{
 				Queue: "queue", PageSize: 25, PageToken: "token",
 				AcceptedLowerBound: &pb.ListRequest_AcceptedAtOrAfterMs{},

service/stovepipe/server/handler/queue_status_test.go

@@ -20,13 +20,49 @@ import (
 
 	"github.com/stretchr/testify/require"
 	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
+	"github.com/uber/submitqueue/platform/errs"
+	"github.com/uber/submitqueue/stovepipe/controller"
+	"github.com/uber/submitqueue/stovepipe/core/queuelog"
+	"github.com/uber/submitqueue/stovepipe/entity"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
 
-func TestQueueStatusContractIsUnimplemented(t *testing.T) {
-	server := &StovepipeServer{}
-	response, err := server.GetQueueStatus(context.Background(), &pb.GetQueueStatusRequest{Queue: "repo/main"})
-	require.Nil(t, response)
-	require.Equal(t, codes.Unimplemented, status.Code(err))
+type fakeQueueStatusController struct {
+	get func(context.Context, entity.GetQueueStatusRequest) (entity.GetQueueStatusResult, error)
+}
+
+func (f fakeQueueStatusController) GetQueueStatus(ctx context.Context, req entity.GetQueueStatusRequest) (entity.GetQueueStatusResult, error) {
+	return f.get(ctx, req)
+}
+
+func TestGetQueueStatus(t *testing.T) {
+	for _, tt := range []struct {
+		err  error
+		code codes.Code
+	}{
+		{nil, codes.OK}, {controller.ErrInvalidRequest, codes.InvalidArgument},
+		{controller.ErrQueueStateNotFound, codes.NotFound}, {controller.ErrCommitStateUnresolved, codes.FailedPrecondition},
+		{queuelog.ErrInconsistentHistory, codes.Internal}, {context.Canceled, codes.Canceled}, {context.DeadlineExceeded, codes.DeadlineExceeded},
+	} {
+		wantErr := tt.err
+		if wantErr == controller.ErrQueueStateNotFound || wantErr == controller.ErrCommitStateUnresolved {
+			wantErr = errs.NewUserError(wantErr)
+		}
+		fake := fakeQueueStatusController{get: func(ctx context.Context, req entity.GetQueueStatusRequest) (entity.GetQueueStatusResult, error) {
+			require.Equal(t, entity.GetQueueStatusRequest{Queue: "repo/main", ChangeURI: "C"}, req)
+			return entity.GetQueueStatusResult{Queue: "repo/main", Enablement: entity.QueueState{Version: 1, State: entity.QueueEnablementDisabled}}, wantErr
+		}}
+		server := NewStovepipeServer(nil, nil, nil, nil, nil, fake)
+		response, err := server.GetQueueStatus(context.Background(), &pb.GetQueueStatusRequest{Queue: "repo/main", ChangeUri: "C"})
+		if wantErr != nil {
+			require.Equal(t, tt.code, status.Code(err))
+			require.Nil(t, response)
+		} else {
+			require.NoError(t, err)
+			require.Equal(t, "repo/main", response.Queue)
+			require.Equal(t, int64(1), response.Enablement.Version)
+			require.Equal(t, pb.QueueEnablement_QUEUE_ENABLEMENT_DISABLED, response.Enablement.State)
+		}
+	}
 }

service/stovepipe/server/handler/server.go

@@ -17,6 +17,11 @@ package handler
 
 import (
 	"context"
+	"errors"
+
+	"github.com/uber/submitqueue/stovepipe/core/queuelog"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
 	"github.com/uber/submitqueue/service/stovepipe/server/mapper"
@@ -31,6 +36,7 @@ type StovepipeServer struct {
 	requestHistoryController controller.RequestHistoryController
 	projectStatusController  controller.ProjectStatusController
 	listController           controller.ListController
+	queueStatusController    controller.GetQueueStatusController
 }
 
 // NewStovepipeServer creates a gRPC service handler from Stovepipe controllers.
@@ -40,13 +46,15 @@ func NewStovepipeServer(
 	requestHistoryController controller.RequestHistoryController,
 	projectStatusController controller.ProjectStatusController,
 	listController controller.ListController,
+	queueStatusController controller.GetQueueStatusController,
 ) *StovepipeServer {
 	return &StovepipeServer{
 		pingController:           pingController,
 		ingestController:         ingestController,
 		requestHistoryController: requestHistoryController,
 		projectStatusController:  projectStatusController,
 		listController:           listController,
+		queueStatusController:    queueStatusController,
 	}
 }
 
@@ -109,3 +117,24 @@ func (s *StovepipeServer) List(ctx context.Context, req *pb.ListRequest) (*pb.Li
 	}
 	return mapper.ListResultToProto(result), nil
 }
+
+func (s *StovepipeServer) GetQueueStatus(ctx context.Context, req *pb.GetQueueStatusRequest) (*pb.GetQueueStatusResponse, error) {
+	result, err := s.queueStatusController.GetQueueStatus(ctx, mapper.ProtoToGetQueueStatusRequest(req))
+	if err != nil {
+		switch {
+		case controller.IsInvalidRequest(err):
+			return nil, status.Error(codes.InvalidArgument, err.Error())
+		case errors.Is(err, controller.ErrQueueStateNotFound):
+			return nil, status.Error(codes.NotFound, err.Error())
+		case errors.Is(err, controller.ErrCommitStateUnresolved):
+			return nil, status.Error(codes.FailedPrecondition, err.Error())
+		case errors.Is(err, queuelog.ErrInconsistentHistory):
+			return nil, status.Error(codes.Internal, err.Error())
+		case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
+			return nil, status.FromContextError(err).Err()
+		default:
+			return nil, err
+		}
+	}
+	return mapper.GetQueueStatusResultToProto(result), nil
+}

service/stovepipe/server/main.go

@@ -295,7 +295,8 @@ func run() error {
 	requestHistoryController := controller.NewRequestHistoryController(logger.Sugar(), scope, storageFty)
 	projectStatusController := controller.NewProjectStatusController(logger.Sugar(), scope, storageFty)
 	listController := controller.NewListController(logger.Sugar(), scope, storageFty, tenants)
-	srv := handler.NewStovepipeServer(pingController, ingestController, requestHistoryController, projectStatusController, listController)
+	queueStatusController := controller.NewGetQueueStatusController(scope, storageFty, sourceControl)
+	srv := handler.NewStovepipeServer(pingController, ingestController, requestHistoryController, projectStatusController, listController, queueStatusController)
 	pb.RegisterStovepipeServer(grpcServer, srv)
 
 	// Register reflection service for debugging with grpcurl

service/stovepipe/server/mapper/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "ingest.go",
         "list.go",
         "project_status.go",
+        "queue_status.go",
         "request_history.go",
     ],
     importpath = "github.com/uber/submitqueue/service/stovepipe/server/mapper",
@@ -22,6 +23,7 @@ go_test(
         "ingest_test.go",
         "list_test.go",
         "project_status_test.go",
+        "queue_status_test.go",
         "request_history_test.go",
     ],
     embed = [":go_default_library"],

service/stovepipe/server/mapper/queue_status.go

@@ -0,0 +1,41 @@
+// 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 mapper
+
+import (
+	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
+	"github.com/uber/submitqueue/stovepipe/entity"
+)
+
+func ProtoToGetQueueStatusRequest(req *pb.GetQueueStatusRequest) entity.GetQueueStatusRequest {
+	return entity.GetQueueStatusRequest{Queue: req.GetQueue(), ChangeURI: req.GetChangeUri()}
+}
+
+func GetQueueStatusResultToProto(result entity.GetQueueStatusResult) *pb.GetQueueStatusResponse {
+	return &pb.GetQueueStatusResponse{
+		Queue: result.Queue, Enablement: queueEnablementStatusToProto(result.Enablement),
+	}
+}
+
+func queueEnablementStatusToProto(applied entity.QueueState) *pb.QueueEnablementStatus {
+	state := pb.QueueEnablement_QUEUE_ENABLEMENT_UNKNOWN
+	switch applied.State {
+	case entity.QueueEnablementEnabled:
+		state = pb.QueueEnablement_QUEUE_ENABLEMENT_ENABLED
+	case entity.QueueEnablementDisabled:
+		state = pb.QueueEnablement_QUEUE_ENABLEMENT_DISABLED
+	}
+	return &pb.QueueEnablementStatus{Version: applied.Version, State: state, EffectiveFromCommitUri: applied.EffectiveFromCommitURI, ChangedAtMs: applied.ChangedAtMs}
+}

service/stovepipe/server/mapper/queue_status_test.go

@@ -0,0 +1,47 @@
+// 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 mapper
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
+	"github.com/uber/submitqueue/stovepipe/entity"
+)
+
+func TestQueueStatusSelectedEnablement(t *testing.T) {
+	for _, tt := range []struct {
+		name      string
+		policy    entity.QueueEnablement
+		wantState pb.QueueEnablement
+	}{
+		{"enabled", entity.QueueEnablementEnabled, pb.QueueEnablement_QUEUE_ENABLEMENT_ENABLED},
+		{"disabled", entity.QueueEnablementDisabled, pb.QueueEnablement_QUEUE_ENABLEMENT_DISABLED},
+		{"unknown values stay unknown", entity.QueueEnablementUnknown, pb.QueueEnablement_QUEUE_ENABLEMENT_UNKNOWN},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			policy := entity.QueueState{Version: 3, State: tt.policy, EffectiveFromCommitURI: "A", ChangedAtMs: 1000}
+			result := GetQueueStatusResultToProto(entity.GetQueueStatusResult{Queue: "repo/main", Enablement: policy})
+			require.Equal(t, "repo/main", result.Queue)
+			require.Equal(t, tt.wantState, result.Enablement.State)
+			require.Equal(t, int64(3), result.Enablement.Version)
+			require.Equal(t, "A", result.Enablement.EffectiveFromCommitUri)
+			require.Equal(t, int64(1000), result.Enablement.ChangedAtMs)
+		})
+	}
+	require.Equal(t, entity.GetQueueStatusRequest{}, ProtoToGetQueueStatusRequest(nil))
+	require.Equal(t, entity.GetQueueStatusRequest{Queue: "repo/main", ChangeURI: "C"}, ProtoToGetQueueStatusRequest(&pb.GetQueueStatusRequest{Queue: "repo/main", ChangeUri: "C"}))
+}

stovepipe/controller/BUILD.bazel

@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "go_default_library",
     srcs = [
+        "get_queue_status.go",
         "ingest.go",
         "list.go",
         "list_pagination.go",
@@ -22,6 +23,7 @@ go_library(
         "//platform/metrics:go_default_library",
         "//platform/publish:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
+        "//stovepipe/core/queuelog:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/sourcecontrol:go_default_library",
@@ -34,6 +36,7 @@ go_library(
 go_test(
     name = "go_default_test",
     srcs = [
+        "get_queue_status_test.go",
         "ingest_test.go",
         "list_consistency_test.go",
         "list_pagination_test.go",
@@ -53,10 +56,12 @@ go_test(
         "//platform/extension/messagequeue/mock:go_default_library",
         "//platform/metrics:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
+        "//stovepipe/core/queuelog:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/core/requestlog/mock:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/sourcecontrol:go_default_library",
+        "//stovepipe/extension/sourcecontrol/fake:go_default_library",
         "//stovepipe/extension/sourcecontrol/mock:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
         "//stovepipe/extension/storage/mock:go_default_library",