feat(stovepipe): add thin Ingest controller and wire it into the example server

Add the IngestController, the pipeline's entry point. It validates the queue name, mints a request ID via the counter extension (format `request/<queue>/<counter>`, following SubmitQueue's convention), builds the resulting `Request` (state `accepted`), logs it, and returns the ID.

This is intentionally thin: resolving the commit URI via the SourceControl extension, persisting the Request, and publishing it onto the pipeline are follow-ups. The example server wires the controller behind the new Ingest RPC using a minimal in-process counter (a real deployment supplies a persistent counter implementation).

Author: behinddwalls

example/stovepipe/server/main.go

@@ -33,17 +33,43 @@ import (
 	"google.golang.org/grpc/reflection"
 )
 
-// StovepipeServer wraps the controller and implements the gRPC service interface.
+// StovepipeServer wraps the controllers and implements the gRPC service interface.
 type StovepipeServer struct {
 	pb.UnimplementedStovepipeServer
-	pingController *controller.PingController
+	pingController   *controller.PingController
+	ingestController *controller.IngestController
 }
 
 // Ping delegates to the controller.
 func (s *StovepipeServer) Ping(ctx context.Context, req *pb.PingRequest) (*pb.PingResponse, error) {
 	return s.pingController.Ping(ctx, req)
 }
 
+// Ingest delegates to the controller.
+func (s *StovepipeServer) Ingest(ctx context.Context, req *pb.IngestRequest) (*pb.IngestResponse, error) {
+	return s.ingestController.Ingest(ctx, req)
+}
+
+// inMemoryCounter is a minimal, process-local counter.Counter used to wire the example
+// server. It is not durable; a real deployment supplies a persistent implementation
+// (e.g. platform/extension/counter/mysql).
+type inMemoryCounter struct {
+	mu     sync.Mutex
+	values map[string]int64
+}
+
+func newInMemoryCounter() *inMemoryCounter {
+	return &inMemoryCounter{values: make(map[string]int64)}
+}
+
+// Next returns the next value in the sequence for the given domain, starting at 1.
+func (c *inMemoryCounter) Next(_ context.Context, domain string) (int64, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.values[domain]++
+	return c.values[domain], nil
+}
+
 func main() {
 	code := 0
 	if err := run(); err != nil {
@@ -108,10 +134,12 @@ func run() error {
 	// Create gRPC server
 	grpcServer := grpc.NewServer()
 
-	// Create ping controller and wrap it for gRPC
+	// Create controllers and wrap them for gRPC
 	pingController := controller.NewPingController(logger, scope)
+	ingestController := controller.NewIngestController(logger.Sugar(), scope, newInMemoryCounter())
 	srv := &StovepipeServer{
-		pingController: pingController,
+		pingController:   pingController,
+		ingestController: ingestController,
 	}
 	pb.RegisterStovepipeServer(grpcServer, srv)
 

stovepipe/controller/BUILD.bazel

@@ -2,26 +2,37 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "controller",
-    srcs = ["ping.go"],
+    srcs = [
+        "ingest.go",
+        "ping.go",
+    ],
     importpath = "github.com/uber/submitqueue/stovepipe/controller",
     visibility = ["//visibility:public"],
     deps = [
         "//api/stovepipe/protopb",
+        "//platform/errs",
+        "//platform/extension/counter",
         "//platform/metrics",
+        "//stovepipe/entity",
         "@com_github_uber_go_tally//:tally",
         "@org_uber_go_zap//:zap",
     ],
 )
 
 go_test(
     name = "controller_test",
-    srcs = ["ping_test.go"],
+    srcs = [
+        "ingest_test.go",
+        "ping_test.go",
+    ],
     embed = [":controller"],
     deps = [
         "//api/stovepipe/protopb",
+        "//platform/extension/counter/mock",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
         "@com_github_uber_go_tally//:tally",
+        "@org_uber_go_mock//gomock",
         "@org_uber_go_zap//:zap",
     ],
 )

stovepipe/controller/ingest.go

@@ -0,0 +1,94 @@
+// Copyright (c) 2025 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"
+	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
+	"github.com/uber/submitqueue/platform/errs"
+	"github.com/uber/submitqueue/platform/extension/counter"
+	"github.com/uber/submitqueue/platform/metrics"
+	"github.com/uber/submitqueue/stovepipe/entity"
+	"go.uber.org/zap"
+)
+
+// ErrInvalidRequest is returned when the request fails validation.
+// This error should be mapped to codes.InvalidArgument at the gRPC layer.
+var ErrInvalidRequest = errs.NewUserError(errors.New("invalid request"))
+
+// IsInvalidRequest returns true if any error in the error chain is ErrInvalidRequest.
+func IsInvalidRequest(err error) bool {
+	return errors.Is(err, ErrInvalidRequest)
+}
+
+// IngestController handles ingest business logic for stovepipe: it admits a queue's newly
+// observed commit into the validation pipeline.
+//
+// This is the thin entry point. It mints a request ID namespaced by the queue and records the
+// resulting Request. Resolving the commit URI via the SourceControl extension, persisting the
+// Request, and publishing it onto the pipeline are deliberately out of scope for now.
+type IngestController struct {
+	logger       *zap.SugaredLogger
+	metricsScope tally.Scope
+	counter      counter.Counter
+}
+
+// NewIngestController creates a new instance of the stovepipe ingest controller.
+func NewIngestController(logger *zap.SugaredLogger, scope tally.Scope, counter counter.Counter) *IngestController {
+	return &IngestController{
+		logger:       logger,
+		metricsScope: scope.SubScope("ingest_controller"),
+		counter:      counter,
+	}
+}
+
+// Ingest admits a queue's newly observed commit into the validation pipeline and returns the minted request ID.
+func (c *IngestController) Ingest(ctx context.Context, req *pb.IngestRequest) (resp *pb.IngestResponse, retErr error) {
+	const opName = "ingest"
+
+	op := metrics.Begin(c.metricsScope, opName)
+	defer func() { op.Complete(retErr) }()
+
+	if req.Queue == "" {
+		return nil, fmt.Errorf("IngestController requires the request to have a queue name specified: %w", ErrInvalidRequest)
+	}
+
+	queue := req.Queue
+
+	// Generate a globally unique request ID namespaced by the queue.
+	seq, err := c.counter.Next(ctx, "request/"+queue)
+	if err != nil {
+		return nil, fmt.Errorf("IngestController failed to generate request ID for queue=%s: %w", queue, err)
+	}
+
+	request := entity.Request{
+		ID:      fmt.Sprintf("request/%s/%d", queue, seq),
+		Queue:   queue,
+		State:   entity.RequestStateAccepted,
+		Version: 1,
+	}
+
+	c.logger.Infow("accepted request",
+		"id", request.ID,
+		"queue", request.Queue,
+		"state", request.State,
+	)
+
+	return &pb.IngestResponse{Id: request.ID}, nil
+}

stovepipe/controller/ingest_test.go

@@ -0,0 +1,70 @@
+// Copyright (c) 2025 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"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/uber-go/tally"
+	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
+	countermock "github.com/uber/submitqueue/platform/extension/counter/mock"
+	"go.uber.org/mock/gomock"
+	"go.uber.org/zap"
+)
+
+func newIngestController(t *testing.T, c *countermock.MockCounter) *IngestController {
+	t.Helper()
+	return NewIngestController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), c)
+}
+
+func TestIngestController_Ingest(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	mockCounter := countermock.NewMockCounter(ctrl)
+	mockCounter.EXPECT().Next(gomock.Any(), "request/monorepo/main").Return(int64(7), nil)
+
+	c := newIngestController(t, mockCounter)
+
+	resp, err := c.Ingest(context.Background(), &pb.IngestRequest{Queue: "monorepo/main"})
+	require.NoError(t, err)
+	assert.Equal(t, "request/monorepo/main/7", resp.Id)
+}
+
+func TestIngestController_Ingest_EmptyQueue(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	mockCounter := countermock.NewMockCounter(ctrl)
+	// Counter must not be consulted when the queue is missing.
+
+	c := newIngestController(t, mockCounter)
+
+	_, err := c.Ingest(context.Background(), &pb.IngestRequest{Queue: ""})
+	require.Error(t, err)
+	assert.True(t, IsInvalidRequest(err))
+}
+
+func TestIngestController_Ingest_CounterError(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	mockCounter := countermock.NewMockCounter(ctrl)
+	mockCounter.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(0), errors.New("counter unavailable"))
+
+	c := newIngestController(t, mockCounter)
+
+	_, err := c.Ingest(context.Background(), &pb.IngestRequest{Queue: "monorepo/main"})
+	require.Error(t, err)
+	assert.False(t, IsInvalidRequest(err))
+}