diff --git a/service/stovepipe/README.md b/service/stovepipe/README.md index 8b66bec6c..2bd7b0b44 100644 --- a/service/stovepipe/README.md +++ b/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). diff --git a/service/stovepipe/server/handler/BUILD.bazel b/service/stovepipe/server/handler/BUILD.bazel index fd1b8a053..7e4b8f946 100644 --- a/service/stovepipe/server/handler/BUILD.bazel +++ b/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", diff --git a/service/stovepipe/server/handler/list_test.go b/service/stovepipe/server/handler/list_test.go index 76e195c5f..2b27b7001 100644 --- a/service/stovepipe/server/handler/list_test.go +++ b/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{}, diff --git a/service/stovepipe/server/handler/queue_status_test.go b/service/stovepipe/server/handler/queue_status_test.go index 48bb10398..5efcd8d98 100644 --- a/service/stovepipe/server/handler/queue_status_test.go +++ b/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) + } + } } diff --git a/service/stovepipe/server/handler/server.go b/service/stovepipe/server/handler/server.go index e7f55660c..d34639d3c 100644 --- a/service/stovepipe/server/handler/server.go +++ b/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,6 +46,7 @@ func NewStovepipeServer( requestHistoryController controller.RequestHistoryController, projectStatusController controller.ProjectStatusController, listController controller.ListController, + queueStatusController controller.GetQueueStatusController, ) *StovepipeServer { return &StovepipeServer{ pingController: pingController, @@ -47,6 +54,7 @@ func NewStovepipeServer( 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 +} diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index b59a3ece2..b19596252 100644 --- a/service/stovepipe/server/main.go +++ b/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 diff --git a/service/stovepipe/server/mapper/BUILD.bazel b/service/stovepipe/server/mapper/BUILD.bazel index dbe7f1baa..0f4c1b5a4 100644 --- a/service/stovepipe/server/mapper/BUILD.bazel +++ b/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"], diff --git a/service/stovepipe/server/mapper/queue_status.go b/service/stovepipe/server/mapper/queue_status.go new file mode 100644 index 000000000..ada1236ff --- /dev/null +++ b/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} +} diff --git a/service/stovepipe/server/mapper/queue_status_test.go b/service/stovepipe/server/mapper/queue_status_test.go new file mode 100644 index 000000000..1605de237 --- /dev/null +++ b/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"})) +} diff --git a/stovepipe/controller/BUILD.bazel b/stovepipe/controller/BUILD.bazel index 7988c25cb..cd1d37c29 100644 --- a/stovepipe/controller/BUILD.bazel +++ b/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", diff --git a/stovepipe/controller/get_queue_status.go b/stovepipe/controller/get_queue_status.go new file mode 100644 index 000000000..cfab1e8ae --- /dev/null +++ b/stovepipe/controller/get_queue_status.go @@ -0,0 +1,157 @@ +// 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 controller + +import ( + "context" + "errors" + "fmt" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/errs" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/core/queuelog" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +var ( + // ErrQueueStateNotFound means the queue has no materialized summary. + ErrQueueStateNotFound = errors.New("queue state not found") + // ErrCommitStateUnresolved means the commit or the retained state lineage cannot be resolved. + ErrCommitStateUnresolved = errors.New("commit state unresolved") +) + +type GetQueueStatusController interface { + GetQueueStatus(context.Context, entity.GetQueueStatusRequest) (entity.GetQueueStatusResult, error) +} + +type getQueueStatusController struct { + scope tally.Scope + stores storage.Factory + sourceControl sourcecontrol.Factory +} + +func NewGetQueueStatusController(scope tally.Scope, stores storage.Factory, sourceControl sourcecontrol.Factory) GetQueueStatusController { + return &getQueueStatusController{scope: scope.SubScope("queue_status_controller"), stores: stores, sourceControl: sourceControl} +} + +func (c *getQueueStatusController) GetQueueStatus(ctx context.Context, req entity.GetQueueStatusRequest) (result entity.GetQueueStatusResult, retErr error) { + op := metrics.Begin(c.scope, "get_queue_status", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...) + defer func() { op.Complete(retErr) }() + if err := validateHistoryIdentifier("queue", req.Queue); err != nil { + return result, err + } + if req.ChangeURI != "" { + if err := validateHistoryIdentifier("change_uri", req.ChangeURI); err != nil { + return result, err + } + } + stores, err := c.stores.For(storage.Config{QueueName: req.Queue}) + if err != nil { + return result, fmt.Errorf("resolve queue state storage: %w", err) + } + store := stores.GetQueueLogStore() + current, err := queuelog.LoadSummary(ctx, stores, req.Queue) + if err != nil { + if storage.IsNotFound(err) { + return result, errs.NewUserError(fmt.Errorf("queue summary %q: %w", req.Queue, ErrQueueStateNotFound)) + } + return result, err + } + result.Queue, result.Enablement = req.Queue, current + if req.ChangeURI != "" { + result.Enablement, err = c.resolveCommitEnablement(ctx, req, store, current) + if err != nil { + return entity.GetQueueStatusResult{}, err + } + } + if ctx.Err() != nil { + return entity.GetQueueStatusResult{}, ctx.Err() + } + return result, nil +} + +func (c *getQueueStatusController) resolveCommitEnablement(ctx context.Context, req entity.GetQueueStatusRequest, store storage.QueueLogStore, current entity.QueueState) (entity.QueueState, error) { + sc, err := c.sourceControl.For(sourcecontrol.Config{QueueName: req.Queue}) + if err != nil { + return entity.QueueState{}, classifyCommitStateError(err) + } + tip, err := sc.Latest(ctx) + if err != nil { + return entity.QueueState{}, classifyCommitStateError(err) + } + belongs, err := sc.IsAncestor(ctx, req.ChangeURI, tip) + if err != nil { + return entity.QueueState{}, classifyCommitStateError(err) + } + if !belongs { + return entity.QueueState{}, errs.NewUserError(ErrCommitStateUnresolved) + } + // A rewritten tip must not turn a formerly enabled commit into initial DISABLED. + if current.EffectiveFromCommitURI != "" { + belongs, err = sc.IsAncestor(ctx, current.EffectiveFromCommitURI, tip) + if err != nil { + return entity.QueueState{}, classifyCommitStateError(err) + } + if !belongs { + return entity.QueueState{}, errs.NewUserError(ErrCommitStateUnresolved) + } + } + for { + if current.Version == 1 { + return current, nil + } + covered, err := sc.IsAncestor(ctx, current.EffectiveFromCommitURI, req.ChangeURI) + if err != nil { + return entity.QueueState{}, classifyCommitStateError(err) + } + if covered { + return current, nil + } + beforeBoundary, err := sc.IsAncestor(ctx, req.ChangeURI, current.EffectiveFromCommitURI) + if err != nil { + return entity.QueueState{}, classifyCommitStateError(err) + } + if !beforeBoundary { + return entity.QueueState{}, errs.NewUserError(ErrCommitStateUnresolved) + } + previous, err := queuelog.LoadVersion(ctx, store, req.Queue, current.Version-1) + if err != nil { + return entity.QueueState{}, err + } + if previous.State == current.State { + return entity.QueueState{}, queuelog.ErrInconsistentHistory + } + if previous.EffectiveFromCommitURI != "" { + forward, err := sc.IsAncestor(ctx, previous.EffectiveFromCommitURI, current.EffectiveFromCommitURI) + if err != nil { + return entity.QueueState{}, classifyCommitStateError(err) + } + if !forward { + return entity.QueueState{}, queuelog.ErrInconsistentHistory + } + } + current = previous + } +} + +func classifyCommitStateError(err error) error { + if sourcecontrol.IsNotFound(err) { + return errs.NewUserError(fmt.Errorf("%w: %w", ErrCommitStateUnresolved, err)) + } + return fmt.Errorf("resolve commit state: %w", err) +} diff --git a/stovepipe/controller/get_queue_status_test.go b/stovepipe/controller/get_queue_status_test.go new file mode 100644 index 000000000..4ce9ab50e --- /dev/null +++ b/stovepipe/controller/get_queue_status_test.go @@ -0,0 +1,360 @@ +// 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 controller + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + + "github.com/uber/submitqueue/platform/errs" + + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/stovepipe/core/queuelog" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" + scfake "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/fake" + scmock "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/mock" + "github.com/uber/submitqueue/stovepipe/extension/storage" + storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" + "go.uber.org/mock/gomock" +) + +type queueStatusFixture struct { + mockController *gomock.Controller + c *getQueueStatusController + store *storagemock.MockQueueSummaryStore + logStore *storagemock.MockQueueLogStore + scFactory *scmock.MockFactory + current entity.QueueState + logs map[string]entity.QueueLog + stateErr, logGetErr error +} + +func newQueueStatusFixture(t *testing.T) *queueStatusFixture { + t.Helper() + ctrl := gomock.NewController(t) + f := &queueStatusFixture{mockController: ctrl, store: storagemock.NewMockQueueSummaryStore(ctrl), + logStore: storagemock.NewMockQueueLogStore(ctrl), scFactory: scmock.NewMockFactory(ctrl), logs: make(map[string]entity.QueueLog)} + factory := storagemock.NewMockFactory(ctrl) + stores := storagemock.NewMockStorage(ctrl) + factory.EXPECT().For(storage.Config{QueueName: "repo/main"}).Return(stores, nil).AnyTimes() + stores.EXPECT().GetQueueSummaryStore().Return(f.store).AnyTimes() + stores.EXPECT().GetQueueLogStore().Return(f.logStore).AnyTimes() + f.c = NewGetQueueStatusController(tally.NoopScope, factory, f.scFactory).(*getQueueStatusController) + f.addState(entity.QueueEnablementDisabled, "") + f.store.EXPECT().Get(gomock.Any()).DoAndReturn(func(context.Context) (entity.QueueSummary, error) { + if f.stateErr != nil { + return entity.QueueSummary{}, f.stateErr + } + if f.current.Queue == "" { + return entity.QueueSummary{}, storage.ErrNotFound + } + return entity.QueueSummary{Queue: f.current.Queue, State: f.current.State, QueueStateVersion: f.current.Version, EffectiveFromCommitURI: f.current.EffectiveFromCommitURI, StateTimestampMs: f.current.ChangedAtMs, Version: f.current.Version + 100}, nil + }).AnyTimes() + f.logStore.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, id string) (entity.QueueLog, error) { + if f.logGetErr != nil { + return entity.QueueLog{}, f.logGetErr + } + log, ok := f.logs[id] + if !ok { + return entity.QueueLog{}, storage.ErrNotFound + } + return log, nil + }).AnyTimes() + return f +} +func (f *queueStatusFixture) addState(state entity.QueueEnablement, boundary string) { + version := f.current.Version + 1 + f.current = entity.QueueState{Queue: "repo/main", State: state, EffectiveFromCommitURI: boundary, ChangedAtMs: version * 1000, Version: version} + id := entity.QueueLogID(version) + f.logs[id] = entity.QueueLog{ID: id, Queue: f.current.Queue, State: state, EffectiveFromCommitURI: boundary, TimestampMs: f.current.ChangedAtMs, QueueStateVersion: version} +} +func (f *queueStatusFixture) expectLinearSourceControl() { + f.scFactory.EXPECT().For(sourcecontrol.Config{QueueName: "repo/main"}). + Return(scfake.New(sourcecontrol.Config{QueueName: "repo/main"}, []string{"G", "F", "E", "D", "C", "B", "A", "before"}), nil) +} + +func TestGetQueueStatusResolvesIntermediateCommitsAndBoundaries(t *testing.T) { + for _, tt := range []struct { + commit string + revision int64 + state entity.QueueEnablement + }{ + {"before", 1, entity.QueueEnablementDisabled}, + {"A", 2, entity.QueueEnablementEnabled}, {"C", 2, entity.QueueEnablementEnabled}, + {"D", 3, entity.QueueEnablementDisabled}, {"F", 3, entity.QueueEnablementDisabled}, + {"G", 4, entity.QueueEnablementEnabled}, + } { + t.Run(tt.commit, func(t *testing.T) { + f := newQueueStatusFixture(t) + f.addState(entity.QueueEnablementEnabled, "A") + f.addState(entity.QueueEnablementDisabled, "D") + f.addState(entity.QueueEnablementEnabled, "G") + // Ignore history newer than the summary. + f.logs[entity.QueueLogID(5)] = entity.QueueLog{ID: entity.QueueLogID(5), Queue: "repo/main", QueueStateVersion: 5, TimestampMs: 5000, State: entity.QueueEnablementDisabled, EffectiveFromCommitURI: "G"} + f.expectLinearSourceControl() + result, err := f.c.GetQueueStatus(context.Background(), entity.GetQueueStatusRequest{Queue: "repo/main", ChangeURI: tt.commit}) + require.NoError(t, err) + require.Equal(t, "repo/main", result.Queue) + require.Equal(t, f.logs[entity.QueueLogID(tt.revision)].Snapshot(), result.Enablement) + require.Equal(t, tt.state, result.Enablement.State) + }) + } +} + +func TestGetQueueStatusNeverEnabledAndSameBoundary(t *testing.T) { + for _, sameBoundary := range []bool{false, true} { + t.Run(map[bool]string{false: "never enabled", true: "latest revision at equal boundary"}[sameBoundary], func(t *testing.T) { + f := newQueueStatusFixture(t) + if sameBoundary { + f.addState(entity.QueueEnablementEnabled, "A") + f.addState(entity.QueueEnablementDisabled, "A") + } + f.expectLinearSourceControl() + result, err := f.c.GetQueueStatus(context.Background(), entity.GetQueueStatusRequest{Queue: "repo/main", ChangeURI: "C"}) + require.NoError(t, err) + require.Equal(t, f.current, result.Enablement) + require.Equal(t, entity.QueueEnablementDisabled, result.Enablement.State) + }) + } +} + +func TestGetQueueStatusCurrentOnlyNeedsNoSourceControl(t *testing.T) { + f := newQueueStatusFixture(t) + f.addState(entity.QueueEnablementEnabled, "A") + result, err := f.c.GetQueueStatus(context.Background(), entity.GetQueueStatusRequest{Queue: "repo/main"}) + require.NoError(t, err) + require.Equal(t, f.logs[entity.QueueLogID(2)].Snapshot(), result.Enablement) +} + +func TestGetQueueStatusPinsStateAcrossConcurrentUpdate(t *testing.T) { + f := newQueueStatusFixture(t) + f.addState(entity.QueueEnablementEnabled, "A") + sc := scfake.New(sourcecontrol.Config{QueueName: "repo/main"}, []string{"G", "F", "D", "C", "A"}) + f.scFactory.EXPECT().For(gomock.Any()).DoAndReturn(func(sourcecontrol.Config) (sourcecontrol.SourceControl, error) { + f.addState(entity.QueueEnablementDisabled, "D") + return sc, nil + }) + result, err := f.c.GetQueueStatus(context.Background(), entity.GetQueueStatusRequest{Queue: "repo/main", ChangeURI: "F"}) + require.NoError(t, err) + require.Equal(t, f.logs[entity.QueueLogID(2)].Snapshot(), result.Enablement) + require.Equal(t, int64(3), f.current.Version) +} + +func TestGetQueueStatusFailuresNeverReturnDisabled(t *testing.T) { + dependencyErr := errors.New("source control unavailable") + for _, tt := range []struct { + name string + setup func(*queueStatusFixture) + want error + }{ + {"unregistered queue", func(f *queueStatusFixture) { f.current = entity.QueueState{} }, ErrQueueStateNotFound}, + {"conflicting current log", func(f *queueStatusFixture) { + log := f.logs[entity.QueueLogID(1)] + log.TimestampMs++ + f.logs[log.ID] = log + }, queuelog.ErrInconsistentHistory}, + {"backward persisted boundary", func(f *queueStatusFixture) { + f.addState(entity.QueueEnablementEnabled, "G") + f.addState(entity.QueueEnablementDisabled, "D") + f.expectLinearSourceControl() + }, queuelog.ErrInconsistentHistory}, + {"redundant persisted state", func(f *queueStatusFixture) { + f.addState(entity.QueueEnablementDisabled, "G") + f.expectLinearSourceControl() + }, queuelog.ErrInconsistentHistory}, + {"missing predecessor", func(f *queueStatusFixture) { + f.addState(entity.QueueEnablementEnabled, "G") + delete(f.logs, entity.QueueLogID(1)) + f.expectLinearSourceControl() + }, queuelog.ErrInconsistentHistory}, + {"unknown commit", func(f *queueStatusFixture) { f.expectLinearSourceControl() }, ErrCommitStateUnresolved}, + {"source dependency", func(f *queueStatusFixture) { f.scFactory.EXPECT().For(gomock.Any()).Return(nil, dependencyErr) }, dependencyErr}, + {"rewritten policy lineage", func(f *queueStatusFixture) { + f.addState(entity.QueueEnablementEnabled, "old-lineage") + sc := scmock.NewMockSourceControl(f.mockController) + sc.EXPECT().Latest(gomock.Any()).Return("new-tip", nil) + sc.EXPECT().IsAncestor(gomock.Any(), "F", "new-tip").Return(true, nil) + sc.EXPECT().IsAncestor(gomock.Any(), "old-lineage", "new-tip").Return(false, nil) + f.scFactory.EXPECT().For(gomock.Any()).Return(sc, nil) + }, ErrCommitStateUnresolved}, + {"unrelated commit", func(f *queueStatusFixture) { + sc := scmock.NewMockSourceControl(f.mockController) + sc.EXPECT().Latest(gomock.Any()).Return("tip", nil) + sc.EXPECT().IsAncestor(gomock.Any(), "F", "tip").Return(false, nil) + f.scFactory.EXPECT().For(gomock.Any()).Return(sc, nil) + }, ErrCommitStateUnresolved}, + } { + t.Run(tt.name, func(t *testing.T) { + f := newQueueStatusFixture(t) + tt.setup(f) + commit := "F" + if tt.name == "backward persisted boundary" { + commit = "B" + } + if tt.name == "unknown commit" { + commit = "unknown" + } + result, err := f.c.GetQueueStatus(context.Background(), entity.GetQueueStatusRequest{Queue: "repo/main", ChangeURI: commit}) + require.ErrorIs(t, err, tt.want) + require.Equal(t, entity.GetQueueStatusResult{}, result) + }) + } +} + +func TestGetQueueStatusRejectsIncomparableCommits(t *testing.T) { + for _, tt := range []struct { + name string + olderBoundary bool + }{ + {"latest boundary", false}, + {"older boundary", true}, + } { + t.Run(tt.name, func(t *testing.T) { + f := newQueueStatusFixture(t) + f.addState(entity.QueueEnablementEnabled, "A") + if tt.olderBoundary { + f.addState(entity.QueueEnablementDisabled, "D") + } + sc := scmock.NewMockSourceControl(f.mockController) + f.scFactory.EXPECT().For(gomock.Any()).Return(sc, nil) + calls := []any{ + sc.EXPECT().Latest(gomock.Any()).Return("M", nil), + sc.EXPECT().IsAncestor(gomock.Any(), "C", "M").Return(true, nil), + sc.EXPECT().IsAncestor(gomock.Any(), f.current.EffectiveFromCommitURI, "M").Return(true, nil), + } + if tt.olderBoundary { + calls = append(calls, + sc.EXPECT().IsAncestor(gomock.Any(), "D", "C").Return(false, nil), + sc.EXPECT().IsAncestor(gomock.Any(), "C", "D").Return(true, nil), + sc.EXPECT().IsAncestor(gomock.Any(), "A", "D").Return(true, nil), + ) + } + calls = append(calls, + sc.EXPECT().IsAncestor(gomock.Any(), "A", "C").Return(false, nil), + sc.EXPECT().IsAncestor(gomock.Any(), "C", "A").Return(false, nil), + ) + gomock.InOrder(calls...) + result, err := f.c.GetQueueStatus(context.Background(), entity.GetQueueStatusRequest{Queue: "repo/main", ChangeURI: "C"}) + require.ErrorIs(t, err, ErrCommitStateUnresolved) + require.Equal(t, entity.GetQueueStatusResult{}, result) + }) + } +} + +func TestGetQueueStatusAncestryFailuresArePreserved(t *testing.T) { + dependency := errors.New("comparison unavailable") + for _, failAt := range []int{0, 1, 2, 3} { + t.Run([]string{"commit membership", "policy lineage", "commit coverage", "reverse ancestry"}[failAt], func(t *testing.T) { + f := newQueueStatusFixture(t) + f.addState(entity.QueueEnablementEnabled, "A") + sc := scmock.NewMockSourceControl(f.mockController) + f.scFactory.EXPECT().For(gomock.Any()).Return(sc, nil) + sc.EXPECT().Latest(gomock.Any()).Return("tip", nil) + calls := [][2]string{{"F", "tip"}, {"A", "tip"}, {"A", "F"}, {"F", "A"}} + expected := make([]any, 0, failAt+1) + for i := 0; i <= failAt; i++ { + call := sc.EXPECT().IsAncestor(gomock.Any(), calls[i][0], calls[i][1]) + if i == failAt { + call.Return(false, dependency) + } else if i == 2 { + call.Return(false, nil) + } else { + call.Return(true, nil) + } + expected = append(expected, call) + } + gomock.InOrder(expected...) + result, err := f.c.GetQueueStatus(context.Background(), entity.GetQueueStatusRequest{Queue: "repo/main", ChangeURI: "F"}) + require.ErrorIs(t, err, dependency) + require.NotErrorIs(t, err, ErrCommitStateUnresolved) + require.Equal(t, entity.GetQueueStatusResult{}, result) + }) + } +} + +func TestGetQueueStatusValidationAndCancellation(t *testing.T) { + for _, req := range []entity.GetQueueStatusRequest{{}, {Queue: strings.Repeat("x", 256)}, {Queue: "repo/main", ChangeURI: strings.Repeat("x", 256)}} { + f := newQueueStatusFixture(t) + _, err := f.c.GetQueueStatus(context.Background(), req) + require.ErrorIs(t, err, ErrInvalidRequest) + } + for _, cancelErr := range []error{context.Canceled, context.DeadlineExceeded} { + f := newQueueStatusFixture(t) + f.scFactory.EXPECT().For(gomock.Any()).Return(nil, cancelErr) + result, err := f.c.GetQueueStatus(context.Background(), entity.GetQueueStatusRequest{Queue: "repo/main", ChangeURI: "C"}) + require.ErrorIs(t, err, cancelErr) + require.Equal(t, entity.GetQueueStatusResult{}, result) + } +} + +func TestGetQueueStatusCurrentOnlyPreservesCancellation(t *testing.T) { + f := newQueueStatusFixture(t) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + result, err := f.c.GetQueueStatus(ctx, entity.GetQueueStatusRequest{Queue: "repo/main"}) + require.ErrorIs(t, err, context.Canceled) + require.Equal(t, entity.GetQueueStatusResult{}, result) +} + +func TestGetQueueStatusMissingCurrentHistoryDoesNotWrite(t *testing.T) { + f := newQueueStatusFixture(t) + f.addState(entity.QueueEnablementEnabled, "A") + delete(f.logs, entity.QueueLogID(f.current.Version)) + result, err := f.c.GetQueueStatus(context.Background(), entity.GetQueueStatusRequest{Queue: "repo/main"}) + require.ErrorIs(t, err, queuelog.ErrInconsistentHistory) + require.Equal(t, entity.GetQueueStatusResult{}, result) + require.Len(t, f.logs, 1) +} + +func TestGetQueueStatusStorageFailures(t *testing.T) { + failure := errors.New("storage unavailable") + for _, tc := range []struct { + name string + setup func(*queueStatusFixture) + }{ + {"snapshot read", func(f *queueStatusFixture) { f.stateErr = failure }}, + {"history reconciliation", func(f *queueStatusFixture) { f.logGetErr = failure }}, + } { + t.Run(tc.name, func(t *testing.T) { + f := newQueueStatusFixture(t) + tc.setup(f) + result, err := f.c.GetQueueStatus(context.Background(), entity.GetQueueStatusRequest{Queue: "repo/main"}) + require.ErrorIs(t, err, failure) + require.Equal(t, entity.GetQueueStatusResult{}, result) + }) + } +} + +func TestInvalidRequestClassificationMatchesDomainCause(t *testing.T) { + for _, tc := range []struct { + name string + err error + invalid bool + }{ + {"invalid selector", ErrInvalidRequest, true}, + {"wrapped invalid selector", fmt.Errorf("selector: %w", ErrInvalidRequest), true}, + {"unregistered queue", errs.NewUserError(ErrQueueStateNotFound), false}, + {"unresolved commit", errs.NewUserError(ErrCommitStateUnresolved), false}, + {"other user error", errs.NewUserError(errors.New("other")), false}, + {"no error", nil, false}, + } { + t.Run(tc.name, func(t *testing.T) { require.Equal(t, tc.invalid, IsInvalidRequest(tc.err)) }) + } +} diff --git a/stovepipe/controller/read_errors.go b/stovepipe/controller/read_errors.go index 398dfe4b2..8e503c07b 100644 --- a/stovepipe/controller/read_errors.go +++ b/stovepipe/controller/read_errors.go @@ -24,12 +24,15 @@ import ( const maxHistoryIdentifierBytes = 255 +var errInvalidRequestCause = errors.New("invalid request") + // ErrInvalidRequest is returned when a request fails validation. -var ErrInvalidRequest = errs.NewUserError(errors.New("invalid request")) +var ErrInvalidRequest = errs.NewUserError(errInvalidRequestCause) // IsInvalidRequest reports whether err contains an invalid request classification. func IsInvalidRequest(err error) bool { - return errors.Is(err, ErrInvalidRequest) + // Framework user-error wrappers match by category, so match the domain cause. + return errors.Is(err, errInvalidRequestCause) } func validateHistoryIdentifier(name, value string) error { diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index e2a9c140c..3b255e96b 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -11,6 +11,7 @@ go_library( "queue_config.go", "queue_log.go", "queue_state.go", + "queue_status.go", "queue_summary.go", "request.go", "request_acceptance.go", diff --git a/stovepipe/entity/queue_status.go b/stovepipe/entity/queue_status.go new file mode 100644 index 000000000..9bb191c63 --- /dev/null +++ b/stovepipe/entity/queue_status.go @@ -0,0 +1,29 @@ +// 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 entity + +type GetQueueStatusRequest struct { + // Queue is the stable repo-and-ref namespace. + Queue string + // ChangeURI is an immutable commit URI; empty selects current enablement. + ChangeURI string +} + +type GetQueueStatusResult struct { + // Queue is the selected repo-and-ref namespace. + Queue string + // Enablement is current enablement or enablement covering the requested commit. + Enablement QueueState +} diff --git a/test/e2e/stovepipe/BUILD.bazel b/test/e2e/stovepipe/BUILD.bazel index a90281887..9cd77929e 100644 --- a/test/e2e/stovepipe/BUILD.bazel +++ b/test/e2e/stovepipe/BUILD.bazel @@ -6,6 +6,7 @@ go_test( "harness_test.go", "ingest_test.go", "list_test.go", + "queue_status_test.go", "suite_test.go", ], data = [ @@ -22,7 +23,12 @@ go_test( ], deps = [ "//api/stovepipe/protopb:go_default_library", + "//stovepipe/core/queuelog:go_default_library", + "//stovepipe/core/queuestate:go_default_library", "//stovepipe/entity:go_default_library", + "//stovepipe/extension/sourcecontrol:go_default_library", + "//stovepipe/extension/sourcecontrol/fake:go_default_library", + "//stovepipe/extension/storage:go_default_library", "//stovepipe/extension/storage/mysql:go_default_library", "//test/testutil:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", @@ -31,6 +37,8 @@ go_test( "@com_github_stretchr_testify//suite:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_golang_google_grpc//:go_default_library", + "@org_golang_google_grpc//codes:go_default_library", + "@org_golang_google_grpc//status:go_default_library", "@org_golang_google_protobuf//proto:go_default_library", ], ) diff --git a/test/e2e/stovepipe/queue_status_test.go b/test/e2e/stovepipe/queue_status_test.go new file mode 100644 index 000000000..dfd2504f4 --- /dev/null +++ b/test/e2e/stovepipe/queue_status_test.go @@ -0,0 +1,118 @@ +// 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 e2e_test + +import ( + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + pb "github.com/uber/submitqueue/api/stovepipe/protopb" + "github.com/uber/submitqueue/stovepipe/core/queuelog" + "github.com/uber/submitqueue/stovepipe/core/queuestate" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" + scfake "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/fake" + "github.com/uber/submitqueue/stovepipe/extension/storage" + mysqlstorage "github.com/uber/submitqueue/stovepipe/extension/storage/mysql" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +const queueStateTestQueue = "e2e-stovepipe/queue-state" + +type queueStateStorageFactory struct{ backend *mysqlstorage.Storage } + +func (f queueStateStorageFactory) For(cfg storage.Config) (storage.Storage, error) { + return f.backend.For(cfg.QueueName) +} + +type queueStateSourceFactory struct{} + +func (queueStateSourceFactory) For(cfg sourcecontrol.Config) (sourcecontrol.SourceControl, error) { + return scfake.New(cfg, []string{"git://" + cfg.QueueName + "/HEAD"}), nil +} + +func (s *StovepipeE2ESuite) TestQueueStatusWithoutValidation() { + t := s.T() + queue := queueStateTestQueue + uri := "git://" + queue + "/HEAD" + initial, err := s.client.GetQueueStatus(s.ctx, &pb.GetQueueStatusRequest{Queue: queue}) + require.NoError(t, err) + require.NotNil(t, initial.Enablement) + require.Equal(t, pb.QueueEnablement_QUEUE_ENABLEMENT_DISABLED, initial.Enablement.State) + require.Equal(t, int64(1), initial.Enablement.Version) + require.Positive(t, initial.Enablement.ChangedAtMs) + manager := queuestate.NewTransitionController(queueStateStorageFactory{backend: s.appStorage}, queueStateSourceFactory{}, queuelog.NewMaterializer(tally.NoopScope)) + for _, tc := range []struct { + name string + state entity.QueueEnablement + wire pb.QueueEnablement + version int64 + }{ + {"enable", entity.QueueEnablementEnabled, pb.QueueEnablement_QUEUE_ENABLEMENT_ENABLED, 2}, + {"disable", entity.QueueEnablementDisabled, pb.QueueEnablement_QUEUE_ENABLEMENT_DISABLED, 3}, + {"reenable", entity.QueueEnablementEnabled, pb.QueueEnablement_QUEUE_ENABLEMENT_ENABLED, 4}, + } { + s.Run(tc.name, func() { + t := s.T() + change, err := manager.Prepare(s.ctx, queue, tc.state) + require.NoError(t, err) + applied, err := manager.Apply(s.ctx, change) + require.NoError(t, err) + result, err := s.client.GetQueueStatus(s.ctx, &pb.GetQueueStatusRequest{Queue: queue, ChangeUri: uri}) + require.NoError(t, err) + require.NotNil(t, result.Enablement) + require.Equal(t, tc.version, result.Enablement.Version) + require.Equal(t, tc.wire, result.Enablement.State) + require.Equal(t, uri, result.Enablement.EffectiveFromCommitUri) + require.Equal(t, applied.ChangedAtMs, result.Enablement.ChangedAtMs) + }) + } + _, err = s.db.ExecContext(s.ctx, "DELETE FROM queue_log WHERE queue = ? AND log_id = ?", queue, entity.QueueLogID(4)) + require.NoError(t, err) + result, err := s.client.GetQueueStatus(s.ctx, &pb.GetQueueStatusRequest{Queue: queue}) + require.Equal(t, codes.Internal, status.Code(err)) + require.Nil(t, result) + _, err = manager.Initialize(s.ctx, queue) + require.NoError(t, err) + result, err = s.client.GetQueueStatus(s.ctx, &pb.GetQueueStatusRequest{Queue: queue}) + require.NoError(t, err) + require.Equal(t, int64(4), result.Enablement.Version) + bound, err := s.appStorage.For(queue) + require.NoError(t, err) + repaired, err := bound.GetQueueLogStore().Get(s.ctx, entity.QueueLogID(4)) + require.NoError(t, err) + require.Equal(t, entity.QueueEnablementEnabled, repaired.State) + var count int + require.NoError(t, s.db.QueryRowContext(s.ctx, "SELECT COUNT(*) FROM request WHERE queue = ?", queue).Scan(&count)) + require.Zero(t, count) +} + +func (s *StovepipeE2ESuite) TestQueueStatusErrors() { + for _, tc := range []struct { + name string + request *pb.GetQueueStatusRequest + code codes.Code + }{ + {"missing queue", &pb.GetQueueStatusRequest{}, codes.InvalidArgument}, + {"unregistered queue", &pb.GetQueueStatusRequest{Queue: "unregistered"}, codes.NotFound}, + {"unknown commit", &pb.GetQueueStatusRequest{Queue: queueStateTestQueue, ChangeUri: "unknown"}, codes.FailedPrecondition}, + } { + s.Run(tc.name, func() { + response, err := s.client.GetQueueStatus(s.ctx, tc.request) + require.Equal(s.T(), tc.code, status.Code(err)) + require.Nil(s.T(), response) + }) + } +} diff --git a/test/e2e/stovepipe/suite_test.go b/test/e2e/stovepipe/suite_test.go index 79136c2a4..0c1fdd7b2 100644 --- a/test/e2e/stovepipe/suite_test.go +++ b/test/e2e/stovepipe/suite_test.go @@ -64,7 +64,7 @@ func (s *StovepipeE2ESuite) SetupSuite() { s.log = testutil.NewTestLogger(t) t.Setenv("MQ_TENANTS", strings.Join([]string{ "monorepo/main", "monorepo/release", "monorepo/slow?buildrunner-fake=build-slow", - listPaginationQueue, listTimeBoundsQueue, listPrimaryQueue, listOtherQueue, listValidationQueue, + listPaginationQueue, listTimeBoundsQueue, listPrimaryQueue, listOtherQueue, listValidationQueue, queueStateTestQueue, }, ",")) s.log.Logf("Starting Stovepipe e2e test suite using docker-compose")