Skip to content

Expose gateway controllers as interfaces #441

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ubettigole merged 1 commit into from
Jul 24, 2026
Merged

Expose gateway controllers as interfaces #441

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions service/submitqueue/gateway/server/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,12 +52,12 @@ import (
// GatewayServer wraps the controller and implements the gRPC service interface
type GatewayServer struct {
pb.UnimplementedSubmitQueueGatewayServer
pingController *controller.PingController
landController *controller.LandController
cancelController *controller.CancelController
requestSummaryController *controller.RequestSummaryController
listController *controller.ListController
requestHistoryController *controller.RequestHistoryController
pingController controller.PingController
landController controller.LandController
cancelController controller.CancelController
requestSummaryController controller.RequestSummaryController
listController controller.ListController
requestHistoryController controller.RequestHistoryController
}

// Ping delegates to the controller
Expand Down
16 changes: 11 additions & 5 deletions submitqueue/gateway/controller/cancel.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,13 @@ import (
// and may still race a successful merge), publishes a CancelRequest to the cancel topic,
// and returns a response. The orchestrator-side cancel controller performs the actual
// state transitions and emits the terminal RequestStatusCancelled log entry.
type CancelController struct {
type CancelController interface {
Cancel(ctx context.Context, req entity.CancelRequest) error
}

var _ CancelController = (*cancelController)(nil)

type cancelController struct {
logger *zap.SugaredLogger
metricsScope tally.Scope
requestSummaryStore storage.RequestSummaryStore
Expand All @@ -46,8 +52,8 @@ type CancelController struct {
// NewCancelController creates a new instance of the gateway cancel controller.
// The controller writes a RequestStatusCancelling log entry through the shared materializer and
// publishes cancel requests to the topic registered under topickey.TopicKeyCancel.
func NewCancelController(logger *zap.SugaredLogger, scope tally.Scope, store storage.Storage, registry consumer.TopicRegistry) *CancelController {
return &CancelController{
func NewCancelController(logger *zap.SugaredLogger, scope tally.Scope, store storage.Storage, registry consumer.TopicRegistry) CancelController {
return &cancelController{
logger: logger,
metricsScope: scope,
requestSummaryStore: store.GetRequestSummaryStore(),
Expand All @@ -64,7 +70,7 @@ func NewCancelController(logger *zap.SugaredLogger, scope tally.Scope, store sto
// completion before the cancel propagates may still land. The RequestStatusCancelling
// entry written here records the user's intent; the terminal outcome is reflected by a
// later RequestStatusCancelled (orchestrator side) or RequestStatusLanded entry.
func (c *CancelController) Cancel(ctx context.Context, req entity.CancelRequest) (retErr error) {
func (c *cancelController) Cancel(ctx context.Context, req entity.CancelRequest) (retErr error) {
const opName = "cancel"

op := metrics.Begin(c.metricsScope, opName, metrics.StorageLatencyBuckets)
Expand Down Expand Up @@ -113,7 +119,7 @@ func (c *CancelController) Cancel(ctx context.Context, req entity.CancelRequest)
}

// publishToQueue publishes a cancel request to the cancel queue for async processing.
func (c *CancelController) publishToQueue(ctx context.Context, cancelRequest entity.CancelRequest) error {
func (c *cancelController) publishToQueue(ctx context.Context, cancelRequest entity.CancelRequest) error {
payload, err := cancelRequest.ToBytes()
if err != nil {
return fmt.Errorf("failed to serialize cancel request: %w", err)
Expand Down
16 changes: 11 additions & 5 deletions submitqueue/gateway/controller/land.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,13 @@ func IsUnrecognizedQueue(err error) bool {
}

// LandController handles land business logic for the gateway
type LandController struct {
type LandController interface {
Land(ctx context.Context, req entity.LandRequest) (entity.LandResult, error)
}

var _ LandController = (*landController)(nil)

type landController struct {
logger *zap.SugaredLogger
metricsScope tally.Scope
counter counter.Counter
Expand All @@ -75,8 +81,8 @@ type LandController struct {
// NewLandController creates a new instance of the gateway land controller.
// The controller publishes land requests to the topic registered under
// topickey.TopicKeyStart in the registry.
func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counter counter.Counter, store storage.Storage, queueConfigs queueconfig.Store, registry consumer.TopicRegistry) *LandController {
return &LandController{
func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counter counter.Counter, store storage.Storage, queueConfigs queueconfig.Store, registry consumer.TopicRegistry) LandController {
return &landController{
logger: logger,
metricsScope: scope.SubScope("land_controller"),
counter: counter,
Expand All @@ -88,7 +94,7 @@ func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counter cou
}

// Land handles the land request and returns the ID assigned to the accepted request.
func (c *LandController) Land(ctx context.Context, req entity.LandRequest) (result entity.LandResult, retErr error) {
func (c *landController) Land(ctx context.Context, req entity.LandRequest) (result entity.LandResult, retErr error) {
const opName = "land"

op := metrics.Begin(c.metricsScope, opName, metrics.StorageLatencyBuckets)
Expand Down Expand Up @@ -179,7 +185,7 @@ func (c *LandController) Land(ctx context.Context, req entity.LandRequest) (resu
}

// publishToQueue publishes a land request to the request queue for async processing.
func (c *LandController) publishToQueue(ctx context.Context, landRequest entity.LandRequest) error {
func (c *landController) publishToQueue(ctx context.Context, landRequest entity.LandRequest) error {
// Serialize land request entity to JSON
payload, err := landRequest.ToBytes()
if err != nil {
Expand Down
14 changes: 10 additions & 4 deletions submitqueue/gateway/controller/list.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,16 +45,22 @@ type listPageToken struct {
}

// ListController handles bounded queue receipt-history queries.
type ListController struct {
type ListController interface {
List(ctx context.Context, req entity.ListRequest) (entity.ListResult, error)
}

var _ ListController = (*listController)(nil)

type listController struct {
logger *zap.SugaredLogger
metricsScope tally.Scope
requestQueueSummaryStore storage.RequestQueueSummaryStore
queueConfigs queueconfig.Store
}

// NewListController creates a gateway list controller.
func NewListController(logger *zap.SugaredLogger, scope tally.Scope, requestQueueSummaryStore storage.RequestQueueSummaryStore, queueConfigs queueconfig.Store) *ListController {
return &ListController{
func NewListController(logger *zap.SugaredLogger, scope tally.Scope, requestQueueSummaryStore storage.RequestQueueSummaryStore, queueConfigs queueconfig.Store) ListController {
return &listController{
logger: logger,
metricsScope: scope.SubScope("list_controller"),
requestQueueSummaryStore: requestQueueSummaryStore,
Expand All @@ -63,7 +69,7 @@ func NewListController(logger *zap.SugaredLogger, scope tally.Scope, requestQueu
}

// List returns one page of requests received for a queue in the supplied half-open time range.
func (c *ListController) List(ctx context.Context, req entity.ListRequest) (result entity.ListResult, retErr error) {
func (c *listController) List(ctx context.Context, req entity.ListRequest) (result entity.ListResult, retErr error) {
op := metrics.Begin(c.metricsScope, "list", metrics.StorageLatencyBuckets)
defer func() { op.Complete(retErr) }()

Expand Down
2 changes: 1 addition & 1 deletion submitqueue/gateway/controller/list_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,7 @@ func TestList_Errors(t *testing.T) {
}
}

func newConfiguredListController(ctrl *gomock.Controller, store storage.RequestQueueSummaryStore) *ListController {
func newConfiguredListController(ctrl *gomock.Controller, store storage.RequestQueueSummaryStore) ListController {
queueConfigs := qcmock.NewMockStore(ctrl)
queueConfigs.EXPECT().Get(gomock.Any(), "q").Return(entity.QueueConfig{}, nil)
return NewListController(zap.NewNop().Sugar(), tally.NoopScope, store, queueConfigs)
Expand Down
14 changes: 10 additions & 4 deletions submitqueue/gateway/controller/ping.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,21 +26,27 @@ import (
)

// PingController handles ping business logic for the gateway
type PingController struct {
type PingController interface {
Ping(ctx context.Context, req *pb.PingRequest) (*pb.PingResponse, error)
}

var _ PingController = (*pingController)(nil)

type pingController struct {
logger *zap.Logger
metricsScope tally.Scope
}

// NewPingController creates a new instance of the gateway ping controller
func NewPingController(logger *zap.Logger, scope tally.Scope) *PingController {
return &PingController{
func NewPingController(logger *zap.Logger, scope tally.Scope) PingController {
return &pingController{
logger: logger,
metricsScope: scope,
}
}

// Ping handles the ping request and returns a response
func (c *PingController) Ping(ctx context.Context, req *pb.PingRequest) (resp *pb.PingResponse, retErr error) {
func (c *pingController) Ping(ctx context.Context, req *pb.PingRequest) (resp *pb.PingResponse, retErr error) {
const opName = "ping"

op := metrics.Begin(c.metricsScope, opName, metrics.FastLatencyBuckets)
Expand Down
17 changes: 12 additions & 5 deletions submitqueue/gateway/controller/request_history.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,16 +30,23 @@ import (
)

// RequestHistoryController handles retained request-log history lookups.
type RequestHistoryController struct {
type RequestHistoryController interface {
GetRequestHistoryByID(ctx context.Context, req entity.GetRequestHistoryByIDRequest) ([]entity.RequestLog, error)
GetRequestHistoryByChangeURI(ctx context.Context, req entity.GetRequestHistoryByChangeURIRequest) ([]entity.RequestHistory, error)
}

var _ RequestHistoryController = (*requestHistoryController)(nil)

type requestHistoryController struct {
logger *zap.SugaredLogger
metricsScope tally.Scope
requestLogStore storage.RequestLogStore
requestURIStore storage.RequestURIStore
}

// NewRequestHistoryController creates a gateway request-history controller.
func NewRequestHistoryController(logger *zap.SugaredLogger, scope tally.Scope, requestLogStore storage.RequestLogStore, requestURIStore storage.RequestURIStore) *RequestHistoryController {
return &RequestHistoryController{
func NewRequestHistoryController(logger *zap.SugaredLogger, scope tally.Scope, requestLogStore storage.RequestLogStore, requestURIStore storage.RequestURIStore) RequestHistoryController {
return &requestHistoryController{
logger: logger,
metricsScope: scope.SubScope("request_history_controller"),
requestLogStore: requestLogStore,
Expand All @@ -48,7 +55,7 @@ func NewRequestHistoryController(logger *zap.SugaredLogger, scope tally.Scope, r
}

// GetRequestHistoryByID returns every retained request-log event for one sqid.
func (c *RequestHistoryController) GetRequestHistoryByID(ctx context.Context, req entity.GetRequestHistoryByIDRequest) (logs []entity.RequestLog, retErr error) {
func (c *requestHistoryController) GetRequestHistoryByID(ctx context.Context, req entity.GetRequestHistoryByIDRequest) (logs []entity.RequestLog, retErr error) {
op := metrics.Begin(c.metricsScope, "get_by_id", metrics.StorageLatencyBuckets)
defer func() { op.Complete(retErr) }()

Expand All @@ -72,7 +79,7 @@ func (c *RequestHistoryController) GetRequestHistoryByID(ctx context.Context, re
}

// GetRequestHistoryByChangeURI returns retained histories for an exact pinned change URI.
func (c *RequestHistoryController) GetRequestHistoryByChangeURI(ctx context.Context, req entity.GetRequestHistoryByChangeURIRequest) (result []entity.RequestHistory, retErr error) {
func (c *requestHistoryController) GetRequestHistoryByChangeURI(ctx context.Context, req entity.GetRequestHistoryByChangeURIRequest) (result []entity.RequestHistory, retErr error) {
op := metrics.Begin(c.metricsScope, "get_by_change_uri", metrics.StorageLatencyBuckets)
defer func() { op.Complete(retErr) }()

Expand Down
18 changes: 9 additions & 9 deletions submitqueue/gateway/controller/request_history_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ func TestHistoryErrors(t *testing.T) {
backendErr := fmt.Errorf("backend down")
tests := []struct {
name string
call func(*RequestHistoryController) error
call func(RequestHistoryController) error
setup func(*storagemock.MockRequestLogStore, *storagemock.MockRequestURIStore)
wantInvalid bool
wantNotFound bool
Expand All @@ -95,7 +95,7 @@ func TestHistoryErrors(t *testing.T) {
}{
{
name: "empty sqid",
call: func(c *RequestHistoryController) error {
call: func(c RequestHistoryController) error {
_, err := c.GetRequestHistoryByID(context.Background(), entity.GetRequestHistoryByIDRequest{})
return err
},
Expand All @@ -104,7 +104,7 @@ func TestHistoryErrors(t *testing.T) {
},
{
name: "empty change URI",
call: func(c *RequestHistoryController) error {
call: func(c RequestHistoryController) error {
_, err := c.GetRequestHistoryByChangeURI(context.Background(), entity.GetRequestHistoryByChangeURIRequest{})
return err
},
Expand All @@ -116,7 +116,7 @@ func TestHistoryErrors(t *testing.T) {
setup: func(logStore *storagemock.MockRequestLogStore, _ *storagemock.MockRequestURIStore) {
logStore.EXPECT().List(gomock.Any(), "missing/1").Return(nil, storage.ErrNotFound)
},
call: func(c *RequestHistoryController) error {
call: func(c RequestHistoryController) error {
_, err := c.GetRequestHistoryByID(context.Background(), entity.GetRequestHistoryByIDRequest{ID: "missing/1"})
return err
},
Expand All @@ -128,7 +128,7 @@ func TestHistoryErrors(t *testing.T) {
setup: func(logStore *storagemock.MockRequestLogStore, _ *storagemock.MockRequestURIStore) {
logStore.EXPECT().List(gomock.Any(), "queue/1").Return(nil, backendErr)
},
call: func(c *RequestHistoryController) error {
call: func(c RequestHistoryController) error {
_, err := c.GetRequestHistoryByID(context.Background(), entity.GetRequestHistoryByIDRequest{ID: "queue/1"})
return err
},
Expand All @@ -138,7 +138,7 @@ func TestHistoryErrors(t *testing.T) {
setup: func(_ *storagemock.MockRequestLogStore, uriStore *storagemock.MockRequestURIStore) {
uriStore.EXPECT().ListByURI(gomock.Any(), "uri", 101).Return(nil, nil)
},
call: func(c *RequestHistoryController) error {
call: func(c RequestHistoryController) error {
_, err := c.GetRequestHistoryByChangeURI(context.Background(), entity.GetRequestHistoryByChangeURIRequest{ChangeURI: "uri"})
return err
},
Expand All @@ -150,7 +150,7 @@ func TestHistoryErrors(t *testing.T) {
setup: func(_ *storagemock.MockRequestLogStore, uriStore *storagemock.MockRequestURIStore) {
uriStore.EXPECT().ListByURI(gomock.Any(), "uri", 101).Return(make([]entity.RequestURI, 101), nil)
},
call: func(c *RequestHistoryController) error {
call: func(c RequestHistoryController) error {
_, err := c.GetRequestHistoryByChangeURI(context.Background(), entity.GetRequestHistoryByChangeURIRequest{ChangeURI: "uri"})
return err
},
Expand All @@ -163,7 +163,7 @@ func TestHistoryErrors(t *testing.T) {
uriStore.EXPECT().ListByURI(gomock.Any(), "uri", 101).Return([]entity.RequestURI{{RequestID: "queue/1"}}, nil)
logStore.EXPECT().List(gomock.Any(), "queue/1").Return(nil, storage.ErrNotFound)
},
call: func(c *RequestHistoryController) error {
call: func(c RequestHistoryController) error {
_, err := c.GetRequestHistoryByChangeURI(context.Background(), entity.GetRequestHistoryByChangeURIRequest{ChangeURI: "uri"})
return err
},
Expand All @@ -176,7 +176,7 @@ func TestHistoryErrors(t *testing.T) {
uriStore.EXPECT().ListByURI(gomock.Any(), "uri", 101).Return([]entity.RequestURI{{RequestID: "malformed"}}, nil)
logStore.EXPECT().List(gomock.Any(), "malformed").Return([]entity.RequestLog{{RequestID: "malformed"}}, nil)
},
call: func(c *RequestHistoryController) error {
call: func(c RequestHistoryController) error {
_, err := c.GetRequestHistoryByChangeURI(context.Background(), entity.GetRequestHistoryByChangeURIRequest{ChangeURI: "uri"})
return err
},
Expand Down
17 changes: 12 additions & 5 deletions submitqueue/gateway/controller/request_summary.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,16 +27,23 @@ import (
)

// RequestSummaryController handles materialized request-summary lookups for the gateway.
type RequestSummaryController struct {
type RequestSummaryController interface {
GetRequestSummaryByID(ctx context.Context, req entity.GetRequestSummaryByIDRequest) (entity.RequestSummary, error)
GetRequestSummaryByChangeURI(ctx context.Context, req entity.GetRequestSummaryByChangeURIRequest) ([]entity.RequestSummary, error)
}

var _ RequestSummaryController = (*requestSummaryController)(nil)

type requestSummaryController struct {
logger *zap.SugaredLogger
metricsScope tally.Scope
requestSummaryStore storage.RequestSummaryStore
requestURIStore storage.RequestURIStore
}

// NewRequestSummaryController creates a gateway request-summary controller.
func NewRequestSummaryController(logger *zap.SugaredLogger, scope tally.Scope, requestSummaryStore storage.RequestSummaryStore, requestURIStore storage.RequestURIStore) *RequestSummaryController {
return &RequestSummaryController{
func NewRequestSummaryController(logger *zap.SugaredLogger, scope tally.Scope, requestSummaryStore storage.RequestSummaryStore, requestURIStore storage.RequestURIStore) RequestSummaryController {
return &requestSummaryController{
logger: logger,
metricsScope: scope.SubScope("request_summary_controller"),
requestSummaryStore: requestSummaryStore,
Expand All @@ -45,7 +52,7 @@ func NewRequestSummaryController(logger *zap.SugaredLogger, scope tally.Scope, r
}

// GetRequestSummaryByID returns the current materialized view of one request.
func (c *RequestSummaryController) GetRequestSummaryByID(ctx context.Context, req entity.GetRequestSummaryByIDRequest) (summary entity.RequestSummary, retErr error) {
func (c *requestSummaryController) GetRequestSummaryByID(ctx context.Context, req entity.GetRequestSummaryByIDRequest) (summary entity.RequestSummary, retErr error) {
op := metrics.Begin(c.metricsScope, "get_by_id", metrics.StorageLatencyBuckets)
defer func() { op.Complete(retErr) }()

Expand All @@ -72,7 +79,7 @@ func (c *RequestSummaryController) GetRequestSummaryByID(ctx context.Context, re
}

// GetRequestSummaryByChangeURI returns current materialized views for an exact pinned change URI.
func (c *RequestSummaryController) GetRequestSummaryByChangeURI(ctx context.Context, req entity.GetRequestSummaryByChangeURIRequest) (summaries []entity.RequestSummary, retErr error) {
func (c *requestSummaryController) GetRequestSummaryByChangeURI(ctx context.Context, req entity.GetRequestSummaryByChangeURIRequest) (summaries []entity.RequestSummary, retErr error) {
op := metrics.Begin(c.metricsScope, "get_by_change_uri", metrics.StorageLatencyBuckets)
defer func() { op.Complete(retErr) }()

Expand Down
Loading
Loading