diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index 438ced4c..559e3222 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/service/stovepipe/server/BUILD.bazel @@ -13,6 +13,7 @@ go_library( "//platform/errs/mysql:go_default_library", "//platform/extension/messagequeue:go_default_library", "//platform/extension/messagequeue/mysql:go_default_library", + "//service/stovepipe/server/mapper:go_default_library", "//stovepipe/controller:go_default_library", "//stovepipe/controller/process:go_default_library", "//stovepipe/core/messagequeue:go_default_library", diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 22c53a60..10e32fe4 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -35,6 +35,7 @@ import ( 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/service/stovepipe/server/mapper" "github.com/uber/submitqueue/stovepipe/controller" "github.com/uber/submitqueue/stovepipe/controller/process" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" @@ -59,9 +60,14 @@ func (s *StovepipeServer) Ping(ctx context.Context, req *pb.PingRequest) (*pb.Pi return s.pingController.Ping(ctx, req) } -// Ingest delegates to the controller. +// Ingest maps the wire request to an entity, delegates to the controller, and maps +// the result back to the wire response. func (s *StovepipeServer) Ingest(ctx context.Context, req *pb.IngestRequest) (*pb.IngestResponse, error) { - return s.ingestController.Ingest(ctx, req) + result, err := s.ingestController.Ingest(ctx, mapper.ProtoToIngestRequest(req)) + if err != nil { + return nil, err + } + return mapper.IngestResultToProto(result), nil } // inMemoryCounter is a minimal, process-local counter.Counter used to wire the example diff --git a/service/stovepipe/server/mapper/BUILD.bazel b/service/stovepipe/server/mapper/BUILD.bazel new file mode 100644 index 00000000..88d60b59 --- /dev/null +++ b/service/stovepipe/server/mapper/BUILD.bazel @@ -0,0 +1,23 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["ingest.go"], + importpath = "github.com/uber/submitqueue/service/stovepipe/server/mapper", + visibility = ["//visibility:public"], + deps = [ + "//api/stovepipe/protopb:go_default_library", + "//stovepipe/entity:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["ingest_test.go"], + embed = [":go_default_library"], + deps = [ + "//api/stovepipe/protopb:go_default_library", + "//stovepipe/entity:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + ], +) diff --git a/service/stovepipe/server/mapper/ingest.go b/service/stovepipe/server/mapper/ingest.go new file mode 100644 index 00000000..148fc5d1 --- /dev/null +++ b/service/stovepipe/server/mapper/ingest.go @@ -0,0 +1,38 @@ +// 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 mapper translates stovepipe wire (proto) types to and from the domain +// entities the controllers operate on. Each RPC gets its own file; translation +// lives here so controllers stay proto-free. +package mapper + +import ( + pb "github.com/uber/submitqueue/api/stovepipe/protopb" + "github.com/uber/submitqueue/stovepipe/entity" +) + +// ProtoToIngestRequest maps the wire IngestRequest to the entity.IngestRequest +// the controller operates on. +func ProtoToIngestRequest(req *pb.IngestRequest) entity.IngestRequest { + return entity.IngestRequest{ + Queue: req.GetQueue(), + } +} + +// IngestResultToProto maps the domain result to the wire IngestResponse. +func IngestResultToProto(result entity.IngestResult) *pb.IngestResponse { + return &pb.IngestResponse{ + Id: result.ID, + } +} diff --git a/service/stovepipe/server/mapper/ingest_test.go b/service/stovepipe/server/mapper/ingest_test.go new file mode 100644 index 00000000..2c715f68 --- /dev/null +++ b/service/stovepipe/server/mapper/ingest_test.go @@ -0,0 +1,75 @@ +// 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 mapper + +import ( + "testing" + + "github.com/stretchr/testify/assert" + pb "github.com/uber/submitqueue/api/stovepipe/protopb" + "github.com/uber/submitqueue/stovepipe/entity" +) + +func TestProtoToIngestRequest(t *testing.T) { + tests := []struct { + name string + req *pb.IngestRequest + expected entity.IngestRequest + }{ + { + name: "maps queue", + req: &pb.IngestRequest{Queue: "monorepo/main"}, + expected: entity.IngestRequest{Queue: "monorepo/main"}, + }, + { + name: "empty request yields zero value", + req: &pb.IngestRequest{}, + expected: entity.IngestRequest{}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := ProtoToIngestRequest(tt.req) + assert.Equal(t, tt.expected, got) + }) + } +} + +func TestIngestResultToProto(t *testing.T) { + tests := []struct { + name string + result entity.IngestResult + expected *pb.IngestResponse + }{ + { + name: "maps ID", + result: entity.IngestResult{ID: "request/monorepo/main/7"}, + expected: &pb.IngestResponse{Id: "request/monorepo/main/7"}, + }, + { + name: "zero value result yields empty response", + result: entity.IngestResult{}, + expected: &pb.IngestResponse{}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := IngestResultToProto(tt.result) + assert.Equal(t, tt.expected, got) + }) + } +} diff --git a/stovepipe/controller/ingest.go b/stovepipe/controller/ingest.go index c9c817b4..db97bfd2 100644 --- a/stovepipe/controller/ingest.go +++ b/stovepipe/controller/ingest.go @@ -20,7 +20,6 @@ import ( "fmt" "github.com/uber-go/tally" - pb "github.com/uber/submitqueue/api/stovepipe/protopb" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" @@ -87,14 +86,14 @@ func NewIngestController( // URI mapping committed but the request write failed — completes the missing steps instead of // returning a dangling reference. The (queue, URI) mapping is the dedup gate, so concurrent // ingests of the same head converge on one request. -func (c *IngestController) Ingest(ctx context.Context, req *pb.IngestRequest) (resp *pb.IngestResponse, retErr error) { +func (c *IngestController) Ingest(ctx context.Context, req entity.IngestRequest) (result entity.IngestResult, 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) + return entity.IngestResult{}, fmt.Errorf("IngestController requires the request to have a queue name specified: %w", ErrInvalidRequest) } queue := req.Queue @@ -102,32 +101,32 @@ func (c *IngestController) Ingest(ctx context.Context, req *pb.IngestRequest) (r // An unresolvable queue/ref is a caller error (unknown queue), not infrastructure. sc, err := c.sourceControl.For(sourcecontrol.Config{QueueName: queue}) if err != nil { - return nil, fmt.Errorf("IngestController failed to resolve source control for queue=%s: %w", queue, err) + return entity.IngestResult{}, fmt.Errorf("IngestController failed to resolve source control for queue=%s: %w", queue, err) } uri, err := sc.Latest(ctx) if err != nil { if sourcecontrol.IsNotFound(err) { - return nil, fmt.Errorf("IngestController could not resolve head for queue=%s: %w: %w", queue, err, ErrInvalidRequest) + return entity.IngestResult{}, fmt.Errorf("IngestController could not resolve head for queue=%s: %w: %w", queue, err, ErrInvalidRequest) } - return nil, fmt.Errorf("IngestController failed to resolve head for queue=%s: %w", queue, err) + return entity.IngestResult{}, fmt.Errorf("IngestController failed to resolve head for queue=%s: %w", queue, err) } // The (queue, URI) mapping is the dedup gate and the source of truth for "does this head // have a request id". id, err := c.resolveID(ctx, queue, uri) if err != nil { - return nil, err + return entity.IngestResult{}, err } // Ensure the request row exists, healing a prior partial write where the mapping committed // but the request did not. request, err := c.ensureRequest(ctx, id, queue, uri) if err != nil { - return nil, err + return entity.IngestResult{}, err } if err := c.advanceQueueLatestRequestID(ctx, queue, id); err != nil { - return nil, err + return entity.IngestResult{}, err } // Publish while the request is still pre-pipeline (Accepted). The process consumer is @@ -136,7 +135,7 @@ func (c *IngestController) Ingest(ctx context.Context, req *pb.IngestRequest) (r // process advances the request past Accepted, ingest stops re-publishing. if request.State == entity.RequestStateAccepted { if err := c.publishProcess(ctx, id, queue); err != nil { - return nil, fmt.Errorf("IngestController failed to publish request %s to process: %w", id, err) + return entity.IngestResult{}, fmt.Errorf("IngestController failed to publish request %s to process: %w", id, err) } } @@ -147,7 +146,7 @@ func (c *IngestController) Ingest(ctx context.Context, req *pb.IngestRequest) (r "state", request.State, ) - return &pb.IngestResponse{Id: id}, nil + return entity.IngestResult{ID: id}, nil } // resolveID returns the request id mapped to (queue, URI), minting and claiming a new one if the diff --git a/stovepipe/controller/ingest_test.go b/stovepipe/controller/ingest_test.go index 89b03fd0..18cd66f2 100644 --- a/stovepipe/controller/ingest_test.go +++ b/stovepipe/controller/ingest_test.go @@ -22,7 +22,6 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/uber-go/tally" - pb "github.com/uber/submitqueue/api/stovepipe/protopb" "github.com/uber/submitqueue/platform/consumer" countermock "github.com/uber/submitqueue/platform/extension/counter/mock" mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" @@ -145,7 +144,6 @@ func TestIngestController_Ingest(t *testing.T) { name: "heals when uri mapped but request missing", queue: testQueue, setup: func(m ingestMocks) { - // Prior attempt committed the URI mapping but failed before the request write. expectResolve(m) m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("request/monorepo/main/3", nil) m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{}, storage.ErrNotFound) @@ -242,7 +240,7 @@ func TestIngestController_Ingest(t *testing.T) { c, m := newIngestController(t, ctrl) tt.setup(m) - resp, err := c.Ingest(context.Background(), &pb.IngestRequest{Queue: tt.queue}) + result, err := c.Ingest(context.Background(), entity.IngestRequest{Queue: tt.queue}) if tt.wantErr { require.Error(t, err) @@ -250,7 +248,7 @@ func TestIngestController_Ingest(t *testing.T) { return } require.NoError(t, err) - assert.Equal(t, tt.wantID, resp.Id) + assert.Equal(t, tt.wantID, result.ID) }) } } diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index 84220bda..64862273 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", srcs = [ + "ingest.go", "queue.go", "queue_config.go", "request.go", diff --git a/stovepipe/entity/ingest.go b/stovepipe/entity/ingest.go new file mode 100644 index 00000000..aa1b42aa --- /dev/null +++ b/stovepipe/entity/ingest.go @@ -0,0 +1,29 @@ +// 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 entity + +// IngestRequest represents the validated inputs of an ingest RPC call. +// The controller resolves the queue's head commit and mints a request ID internally. +type IngestRequest struct { + // Queue is the name of the queue whose head commit should be ingested. + Queue string +} + +// IngestResult is the outcome of a successful ingest operation. +type IngestResult struct { + // ID is the globally unique request identifier assigned to the ingested commit. + // Format: "request//". + ID string +}