diff --git a/api/submitqueue/gateway/proto/gateway.proto b/api/submitqueue/gateway/proto/gateway.proto index 244f7b43..45e6ad60 100644 --- a/api/submitqueue/gateway/proto/gateway.proto +++ b/api/submitqueue/gateway/proto/gateway.proto @@ -82,22 +82,46 @@ message CancelRequest { message CancelResponse { } -// StatusRequest defines a request to look up the current status of a previously submitted request. -message StatusRequest { - // Globally unique identifier for the request, as returned by Land in the LandResponse. +// RequestSummary is the gateway-owned materialized current view of one received request. +message RequestSummary { + // Globally unique request identifier. This is request_id in internal storage. string sqid = 1; + // Queue supplied when the request was received. + string queue = 2; + // Change URIs supplied when the request was received, in caller order. + repeated string change_uris = 3; + // Receipt time in Unix milliseconds. + int64 received_at_ms = 4; + // Current customer-friendly status of the request. + string status = 5; + // Last error associated with the current status. Empty when absent. + string last_error = 6; + // Display and debugging metadata associated with the current status. + map metadata = 7; +} + +// GetRequestSummaryByIDRequest selects one request by the sqid returned from Land. +message GetRequestSummaryByIDRequest { + // Globally unique identifier for the request. + string sqid = 1; +} + +// GetRequestSummaryByIDResponse contains the current materialized request view. +message GetRequestSummaryByIDResponse { + // Matching request. + RequestSummary request = 1; +} + +// GetRequestSummaryByChangeURIRequest selects requests by an exact pinned change URI. +message GetRequestSummaryByChangeURIRequest { + // Exact change URI supplied in a Land request. + string change_uri = 1; } -// StatusResponse defines the response to a status request. -// The status is eventually consistent with the request store; it might take some time to converge, typically no more than a few seconds. -message StatusResponse { - // Current customer-friendly status of the request (e.g. "accepted", "validating", "landed", "error"). - // Status is a free-form string because the system may introduce new statuses without breaking existing clients. - string status = 1; - // Last error message associated with the current status. Empty string if there is no error. - string last_error = 2; - // Free-form key-value metadata associated with the current status, for display or debugging purposes. Empty if none. - map metadata = 3; +// GetRequestSummaryByChangeURIResponse contains matching requests newest first. +message GetRequestSummaryByChangeURIResponse { + // Matching requests ordered by receipt time descending, then sqid descending. + repeated RequestSummary requests = 1; } // *************** @@ -124,6 +148,8 @@ message RequestNotFoundError { Error error = 1; // The sqid that was not found. string sqid = 2; + // Exact change URI that was not found. Populated only for change-URI lookup. + string change_uri = 3; } // *************** @@ -149,10 +175,13 @@ service SubmitQueueGateway { // Cancellation is NOT GUARANTEED: a request that has already merged, or that races to completion before the cancel // signal propagates through the pipeline, may still land (or end in an error). Callers must NOT assume that a // successful Cancel response means the request was cancelled — the actual terminal outcome (cancelled, landed, or - // error) must be checked through the separate status / request-log API. + // error) must be checked through the request-summary or request-history APIs. rpc Cancel(CancelRequest) returns (CancelResponse) {} - // Status returns the current status of a previously submitted request, identified by its sqid. - // The status is eventually consistent with the request store and reconciled from the append-only request log. - rpc Status(StatusRequest) returns (StatusResponse) {} + // GetRequestSummaryByID returns the current materialized status of one request. + rpc GetRequestSummaryByID(GetRequestSummaryByIDRequest) returns (GetRequestSummaryByIDResponse) {} + + // GetRequestSummaryByChangeURI returns current materialized statuses for an exact pinned change URI. + rpc GetRequestSummaryByChangeURI(GetRequestSummaryByChangeURIRequest) returns (GetRequestSummaryByChangeURIResponse) {} + } diff --git a/api/submitqueue/gateway/protopb/gateway.pb.go b/api/submitqueue/gateway/protopb/gateway.pb.go index 20c020f3..4ba88ee7 100644 --- a/api/submitqueue/gateway/protopb/gateway.pb.go +++ b/api/submitqueue/gateway/protopb/gateway.pb.go @@ -367,29 +367,41 @@ func (*CancelResponse) Descriptor() ([]byte, []int) { return file_gateway_proto_rawDescGZIP(), []int{5} } -// StatusRequest defines a request to look up the current status of a previously submitted request. -type StatusRequest struct { +// RequestSummary is the gateway-owned materialized current view of one received request. +type RequestSummary struct { state protoimpl.MessageState `protogen:"open.v1"` - // Globally unique identifier for the request, as returned by Land in the LandResponse. - Sqid string `protobuf:"bytes,1,opt,name=sqid,proto3" json:"sqid,omitempty"` + // Globally unique request identifier. This is request_id in internal storage. + Sqid string `protobuf:"bytes,1,opt,name=sqid,proto3" json:"sqid,omitempty"` + // Queue supplied when the request was received. + Queue string `protobuf:"bytes,2,opt,name=queue,proto3" json:"queue,omitempty"` + // Change URIs supplied when the request was received, in caller order. + ChangeUris []string `protobuf:"bytes,3,rep,name=change_uris,json=changeUris,proto3" json:"change_uris,omitempty"` + // Receipt time in Unix milliseconds. + ReceivedAtMs int64 `protobuf:"varint,4,opt,name=received_at_ms,json=receivedAtMs,proto3" json:"received_at_ms,omitempty"` + // Current customer-friendly status of the request. + Status string `protobuf:"bytes,5,opt,name=status,proto3" json:"status,omitempty"` + // Last error associated with the current status. Empty when absent. + LastError string `protobuf:"bytes,6,opt,name=last_error,json=lastError,proto3" json:"last_error,omitempty"` + // Display and debugging metadata associated with the current status. + Metadata map[string]string `protobuf:"bytes,7,rep,name=metadata,proto3" json:"metadata,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } -func (x *StatusRequest) Reset() { - *x = StatusRequest{} +func (x *RequestSummary) Reset() { + *x = RequestSummary{} mi := &file_gateway_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } -func (x *StatusRequest) String() string { +func (x *RequestSummary) String() string { return protoimpl.X.MessageStringOf(x) } -func (*StatusRequest) ProtoMessage() {} +func (*RequestSummary) ProtoMessage() {} -func (x *StatusRequest) ProtoReflect() protoreflect.Message { +func (x *RequestSummary) ProtoReflect() protoreflect.Message { mi := &file_gateway_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -401,47 +413,83 @@ func (x *StatusRequest) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -// Deprecated: Use StatusRequest.ProtoReflect.Descriptor instead. -func (*StatusRequest) Descriptor() ([]byte, []int) { +// Deprecated: Use RequestSummary.ProtoReflect.Descriptor instead. +func (*RequestSummary) Descriptor() ([]byte, []int) { return file_gateway_proto_rawDescGZIP(), []int{6} } -func (x *StatusRequest) GetSqid() string { +func (x *RequestSummary) GetSqid() string { if x != nil { return x.Sqid } return "" } -// StatusResponse defines the response to a status request. -// The status is eventually consistent with the request store; it might take some time to converge, typically no more than a few seconds. -type StatusResponse struct { +func (x *RequestSummary) GetQueue() string { + if x != nil { + return x.Queue + } + return "" +} + +func (x *RequestSummary) GetChangeUris() []string { + if x != nil { + return x.ChangeUris + } + return nil +} + +func (x *RequestSummary) GetReceivedAtMs() int64 { + if x != nil { + return x.ReceivedAtMs + } + return 0 +} + +func (x *RequestSummary) GetStatus() string { + if x != nil { + return x.Status + } + return "" +} + +func (x *RequestSummary) GetLastError() string { + if x != nil { + return x.LastError + } + return "" +} + +func (x *RequestSummary) GetMetadata() map[string]string { + if x != nil { + return x.Metadata + } + return nil +} + +// GetRequestSummaryByIDRequest selects one request by the sqid returned from Land. +type GetRequestSummaryByIDRequest struct { state protoimpl.MessageState `protogen:"open.v1"` - // Current customer-friendly status of the request (e.g. "accepted", "validating", "landed", "error"). - // Status is a free-form string because the system may introduce new statuses without breaking existing clients. - Status string `protobuf:"bytes,1,opt,name=status,proto3" json:"status,omitempty"` - // Last error message associated with the current status. Empty string if there is no error. - LastError string `protobuf:"bytes,2,opt,name=last_error,json=lastError,proto3" json:"last_error,omitempty"` - // Free-form key-value metadata associated with the current status, for display or debugging purposes. Empty if none. - Metadata map[string]string `protobuf:"bytes,3,rep,name=metadata,proto3" json:"metadata,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + // Globally unique identifier for the request. + Sqid string `protobuf:"bytes,1,opt,name=sqid,proto3" json:"sqid,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } -func (x *StatusResponse) Reset() { - *x = StatusResponse{} +func (x *GetRequestSummaryByIDRequest) Reset() { + *x = GetRequestSummaryByIDRequest{} mi := &file_gateway_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } -func (x *StatusResponse) String() string { +func (x *GetRequestSummaryByIDRequest) String() string { return protoimpl.X.MessageStringOf(x) } -func (*StatusResponse) ProtoMessage() {} +func (*GetRequestSummaryByIDRequest) ProtoMessage() {} -func (x *StatusResponse) ProtoReflect() protoreflect.Message { +func (x *GetRequestSummaryByIDRequest) ProtoReflect() protoreflect.Message { mi := &file_gateway_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -453,28 +501,152 @@ func (x *StatusResponse) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -// Deprecated: Use StatusResponse.ProtoReflect.Descriptor instead. -func (*StatusResponse) Descriptor() ([]byte, []int) { +// Deprecated: Use GetRequestSummaryByIDRequest.ProtoReflect.Descriptor instead. +func (*GetRequestSummaryByIDRequest) Descriptor() ([]byte, []int) { return file_gateway_proto_rawDescGZIP(), []int{7} } -func (x *StatusResponse) GetStatus() string { +func (x *GetRequestSummaryByIDRequest) GetSqid() string { if x != nil { - return x.Status + return x.Sqid } return "" } -func (x *StatusResponse) GetLastError() string { +// GetRequestSummaryByIDResponse contains the current materialized request view. +type GetRequestSummaryByIDResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Matching request. + Request *RequestSummary `protobuf:"bytes,1,opt,name=request,proto3" json:"request,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetRequestSummaryByIDResponse) Reset() { + *x = GetRequestSummaryByIDResponse{} + mi := &file_gateway_proto_msgTypes[8] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetRequestSummaryByIDResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetRequestSummaryByIDResponse) ProtoMessage() {} + +func (x *GetRequestSummaryByIDResponse) ProtoReflect() protoreflect.Message { + mi := &file_gateway_proto_msgTypes[8] if x != nil { - return x.LastError + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetRequestSummaryByIDResponse.ProtoReflect.Descriptor instead. +func (*GetRequestSummaryByIDResponse) Descriptor() ([]byte, []int) { + return file_gateway_proto_rawDescGZIP(), []int{8} +} + +func (x *GetRequestSummaryByIDResponse) GetRequest() *RequestSummary { + if x != nil { + return x.Request + } + return nil +} + +// GetRequestSummaryByChangeURIRequest selects requests by an exact pinned change URI. +type GetRequestSummaryByChangeURIRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Exact change URI supplied in a Land request. + ChangeUri string `protobuf:"bytes,1,opt,name=change_uri,json=changeUri,proto3" json:"change_uri,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetRequestSummaryByChangeURIRequest) Reset() { + *x = GetRequestSummaryByChangeURIRequest{} + mi := &file_gateway_proto_msgTypes[9] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetRequestSummaryByChangeURIRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetRequestSummaryByChangeURIRequest) ProtoMessage() {} + +func (x *GetRequestSummaryByChangeURIRequest) ProtoReflect() protoreflect.Message { + mi := &file_gateway_proto_msgTypes[9] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetRequestSummaryByChangeURIRequest.ProtoReflect.Descriptor instead. +func (*GetRequestSummaryByChangeURIRequest) Descriptor() ([]byte, []int) { + return file_gateway_proto_rawDescGZIP(), []int{9} +} + +func (x *GetRequestSummaryByChangeURIRequest) GetChangeUri() string { + if x != nil { + return x.ChangeUri } return "" } -func (x *StatusResponse) GetMetadata() map[string]string { +// GetRequestSummaryByChangeURIResponse contains matching requests newest first. +type GetRequestSummaryByChangeURIResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Matching requests ordered by receipt time descending, then sqid descending. + Requests []*RequestSummary `protobuf:"bytes,1,rep,name=requests,proto3" json:"requests,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetRequestSummaryByChangeURIResponse) Reset() { + *x = GetRequestSummaryByChangeURIResponse{} + mi := &file_gateway_proto_msgTypes[10] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetRequestSummaryByChangeURIResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetRequestSummaryByChangeURIResponse) ProtoMessage() {} + +func (x *GetRequestSummaryByChangeURIResponse) ProtoReflect() protoreflect.Message { + mi := &file_gateway_proto_msgTypes[10] if x != nil { - return x.Metadata + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetRequestSummaryByChangeURIResponse.ProtoReflect.Descriptor instead. +func (*GetRequestSummaryByChangeURIResponse) Descriptor() ([]byte, []int) { + return file_gateway_proto_rawDescGZIP(), []int{10} +} + +func (x *GetRequestSummaryByChangeURIResponse) GetRequests() []*RequestSummary { + if x != nil { + return x.Requests } return nil } @@ -490,7 +662,7 @@ type Error struct { func (x *Error) Reset() { *x = Error{} - mi := &file_gateway_proto_msgTypes[8] + mi := &file_gateway_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -502,7 +674,7 @@ func (x *Error) String() string { func (*Error) ProtoMessage() {} func (x *Error) ProtoReflect() protoreflect.Message { - mi := &file_gateway_proto_msgTypes[8] + mi := &file_gateway_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -515,7 +687,7 @@ func (x *Error) ProtoReflect() protoreflect.Message { // Deprecated: Use Error.ProtoReflect.Descriptor instead. func (*Error) Descriptor() ([]byte, []int) { - return file_gateway_proto_rawDescGZIP(), []int{8} + return file_gateway_proto_rawDescGZIP(), []int{11} } func (x *Error) GetMessage() string { @@ -538,7 +710,7 @@ type UnrecognizedQueueError struct { func (x *UnrecognizedQueueError) Reset() { *x = UnrecognizedQueueError{} - mi := &file_gateway_proto_msgTypes[9] + mi := &file_gateway_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -550,7 +722,7 @@ func (x *UnrecognizedQueueError) String() string { func (*UnrecognizedQueueError) ProtoMessage() {} func (x *UnrecognizedQueueError) ProtoReflect() protoreflect.Message { - mi := &file_gateway_proto_msgTypes[9] + mi := &file_gateway_proto_msgTypes[12] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -563,7 +735,7 @@ func (x *UnrecognizedQueueError) ProtoReflect() protoreflect.Message { // Deprecated: Use UnrecognizedQueueError.ProtoReflect.Descriptor instead. func (*UnrecognizedQueueError) Descriptor() ([]byte, []int) { - return file_gateway_proto_rawDescGZIP(), []int{9} + return file_gateway_proto_rawDescGZIP(), []int{12} } func (x *UnrecognizedQueueError) GetError() *Error { @@ -586,14 +758,16 @@ type RequestNotFoundError struct { // Free text error message describing the error. Error *Error `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"` // The sqid that was not found. - Sqid string `protobuf:"bytes,2,opt,name=sqid,proto3" json:"sqid,omitempty"` + Sqid string `protobuf:"bytes,2,opt,name=sqid,proto3" json:"sqid,omitempty"` + // Exact change URI that was not found. Populated only for change-URI lookup. + ChangeUri string `protobuf:"bytes,3,opt,name=change_uri,json=changeUri,proto3" json:"change_uri,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } func (x *RequestNotFoundError) Reset() { *x = RequestNotFoundError{} - mi := &file_gateway_proto_msgTypes[10] + mi := &file_gateway_proto_msgTypes[13] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -605,7 +779,7 @@ func (x *RequestNotFoundError) String() string { func (*RequestNotFoundError) ProtoMessage() {} func (x *RequestNotFoundError) ProtoReflect() protoreflect.Message { - mi := &file_gateway_proto_msgTypes[10] + mi := &file_gateway_proto_msgTypes[13] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -618,7 +792,7 @@ func (x *RequestNotFoundError) ProtoReflect() protoreflect.Message { // Deprecated: Use RequestNotFoundError.ProtoReflect.Descriptor instead. func (*RequestNotFoundError) Descriptor() ([]byte, []int) { - return file_gateway_proto_rawDescGZIP(), []int{10} + return file_gateway_proto_rawDescGZIP(), []int{13} } func (x *RequestNotFoundError) GetError() *Error { @@ -635,6 +809,13 @@ func (x *RequestNotFoundError) GetSqid() string { return "" } +func (x *RequestNotFoundError) GetChangeUri() string { + if x != nil { + return x.ChangeUri + } + return "" +} + var File_gateway_proto protoreflect.FileDescriptor const file_gateway_proto_rawDesc = "" + @@ -656,30 +837,45 @@ const file_gateway_proto_rawDesc = "" + "\rCancelRequest\x12\x12\n" + "\x04sqid\x18\x01 \x01(\tR\x04sqid\x12\x16\n" + "\x06reason\x18\x02 \x01(\tR\x06reason\"\x10\n" + - "\x0eCancelResponse\"#\n" + - "\rStatusRequest\x12\x12\n" + - "\x04sqid\x18\x01 \x01(\tR\x04sqid\"\xd8\x01\n" + - "\x0eStatusResponse\x12\x16\n" + - "\x06status\x18\x01 \x01(\tR\x06status\x12\x1d\n" + + "\x0eCancelResponse\"\xc9\x02\n" + + "\x0eRequestSummary\x12\x12\n" + + "\x04sqid\x18\x01 \x01(\tR\x04sqid\x12\x14\n" + + "\x05queue\x18\x02 \x01(\tR\x05queue\x12\x1f\n" + + "\vchange_uris\x18\x03 \x03(\tR\n" + + "changeUris\x12$\n" + + "\x0ereceived_at_ms\x18\x04 \x01(\x03R\freceivedAtMs\x12\x16\n" + + "\x06status\x18\x05 \x01(\tR\x06status\x12\x1d\n" + "\n" + - "last_error\x18\x02 \x01(\tR\tlastError\x12R\n" + - "\bmetadata\x18\x03 \x03(\v26.uber.submitqueue.gateway.StatusResponse.MetadataEntryR\bmetadata\x1a;\n" + + "last_error\x18\x06 \x01(\tR\tlastError\x12R\n" + + "\bmetadata\x18\a \x03(\v26.uber.submitqueue.gateway.RequestSummary.MetadataEntryR\bmetadata\x1a;\n" + "\rMetadataEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"!\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"2\n" + + "\x1cGetRequestSummaryByIDRequest\x12\x12\n" + + "\x04sqid\x18\x01 \x01(\tR\x04sqid\"c\n" + + "\x1dGetRequestSummaryByIDResponse\x12B\n" + + "\arequest\x18\x01 \x01(\v2(.uber.submitqueue.gateway.RequestSummaryR\arequest\"D\n" + + "#GetRequestSummaryByChangeURIRequest\x12\x1d\n" + + "\n" + + "change_uri\x18\x01 \x01(\tR\tchangeUri\"l\n" + + "$GetRequestSummaryByChangeURIResponse\x12D\n" + + "\brequests\x18\x01 \x03(\v2(.uber.submitqueue.gateway.RequestSummaryR\brequests\"!\n" + "\x05Error\x12\x18\n" + "\amessage\x18\x01 \x01(\tR\amessage\"e\n" + "\x16UnrecognizedQueueError\x125\n" + "\x05error\x18\x01 \x01(\v2\x1f.uber.submitqueue.gateway.ErrorR\x05error\x12\x14\n" + - "\x05queue\x18\x02 \x01(\tR\x05queue\"a\n" + + "\x05queue\x18\x02 \x01(\tR\x05queue\"\x80\x01\n" + "\x14RequestNotFoundError\x125\n" + "\x05error\x18\x01 \x01(\v2\x1f.uber.submitqueue.gateway.ErrorR\x05error\x12\x12\n" + - "\x04sqid\x18\x02 \x01(\tR\x04sqid2\x84\x03\n" + + "\x04sqid\x18\x02 \x01(\tR\x04sqid\x12\x1d\n" + + "\n" + + "change_uri\x18\x03 \x01(\tR\tchangeUri2\xd4\x04\n" + "\x12SubmitQueueGateway\x12W\n" + "\x04Ping\x12%.uber.submitqueue.gateway.PingRequest\x1a&.uber.submitqueue.gateway.PingResponse\"\x00\x12W\n" + "\x04Land\x12%.uber.submitqueue.gateway.LandRequest\x1a&.uber.submitqueue.gateway.LandResponse\"\x00\x12]\n" + - "\x06Cancel\x12'.uber.submitqueue.gateway.CancelRequest\x1a(.uber.submitqueue.gateway.CancelResponse\"\x00\x12]\n" + - "\x06Status\x12'.uber.submitqueue.gateway.StatusRequest\x1a(.uber.submitqueue.gateway.StatusResponse\"\x00Bk\n" + + "\x06Cancel\x12'.uber.submitqueue.gateway.CancelRequest\x1a(.uber.submitqueue.gateway.CancelResponse\"\x00\x12\x8a\x01\n" + + "\x15GetRequestSummaryByID\x126.uber.submitqueue.gateway.GetRequestSummaryByIDRequest\x1a7.uber.submitqueue.gateway.GetRequestSummaryByIDResponse\"\x00\x12\x9f\x01\n" + + "\x1cGetRequestSummaryByChangeURI\x12=.uber.submitqueue.gateway.GetRequestSummaryByChangeURIRequest\x1a>.uber.submitqueue.gateway.GetRequestSummaryByChangeURIResponse\"\x00Bk\n" + "\x1ccom.uber.submitqueue.gatewayB\fGatewayProtoP\x01Z;github.com/uber/submitqueue/api/submitqueue/gateway/protopbb\x06proto3" var ( @@ -694,42 +890,49 @@ func file_gateway_proto_rawDescGZIP() []byte { return file_gateway_proto_rawDescData } -var file_gateway_proto_msgTypes = make([]protoimpl.MessageInfo, 12) +var file_gateway_proto_msgTypes = make([]protoimpl.MessageInfo, 15) var file_gateway_proto_goTypes = []any{ - (*PingRequest)(nil), // 0: uber.submitqueue.gateway.PingRequest - (*PingResponse)(nil), // 1: uber.submitqueue.gateway.PingResponse - (*LandRequest)(nil), // 2: uber.submitqueue.gateway.LandRequest - (*LandResponse)(nil), // 3: uber.submitqueue.gateway.LandResponse - (*CancelRequest)(nil), // 4: uber.submitqueue.gateway.CancelRequest - (*CancelResponse)(nil), // 5: uber.submitqueue.gateway.CancelResponse - (*StatusRequest)(nil), // 6: uber.submitqueue.gateway.StatusRequest - (*StatusResponse)(nil), // 7: uber.submitqueue.gateway.StatusResponse - (*Error)(nil), // 8: uber.submitqueue.gateway.Error - (*UnrecognizedQueueError)(nil), // 9: uber.submitqueue.gateway.UnrecognizedQueueError - (*RequestNotFoundError)(nil), // 10: uber.submitqueue.gateway.RequestNotFoundError - nil, // 11: uber.submitqueue.gateway.StatusResponse.MetadataEntry - (*protopb.Change)(nil), // 12: uber.base.change.Change - (protopb1.Strategy)(0), // 13: uber.base.mergestrategy.Strategy + (*PingRequest)(nil), // 0: uber.submitqueue.gateway.PingRequest + (*PingResponse)(nil), // 1: uber.submitqueue.gateway.PingResponse + (*LandRequest)(nil), // 2: uber.submitqueue.gateway.LandRequest + (*LandResponse)(nil), // 3: uber.submitqueue.gateway.LandResponse + (*CancelRequest)(nil), // 4: uber.submitqueue.gateway.CancelRequest + (*CancelResponse)(nil), // 5: uber.submitqueue.gateway.CancelResponse + (*RequestSummary)(nil), // 6: uber.submitqueue.gateway.RequestSummary + (*GetRequestSummaryByIDRequest)(nil), // 7: uber.submitqueue.gateway.GetRequestSummaryByIDRequest + (*GetRequestSummaryByIDResponse)(nil), // 8: uber.submitqueue.gateway.GetRequestSummaryByIDResponse + (*GetRequestSummaryByChangeURIRequest)(nil), // 9: uber.submitqueue.gateway.GetRequestSummaryByChangeURIRequest + (*GetRequestSummaryByChangeURIResponse)(nil), // 10: uber.submitqueue.gateway.GetRequestSummaryByChangeURIResponse + (*Error)(nil), // 11: uber.submitqueue.gateway.Error + (*UnrecognizedQueueError)(nil), // 12: uber.submitqueue.gateway.UnrecognizedQueueError + (*RequestNotFoundError)(nil), // 13: uber.submitqueue.gateway.RequestNotFoundError + nil, // 14: uber.submitqueue.gateway.RequestSummary.MetadataEntry + (*protopb.Change)(nil), // 15: uber.base.change.Change + (protopb1.Strategy)(0), // 16: uber.base.mergestrategy.Strategy } var file_gateway_proto_depIdxs = []int32{ - 12, // 0: uber.submitqueue.gateway.LandRequest.change:type_name -> uber.base.change.Change - 13, // 1: uber.submitqueue.gateway.LandRequest.strategy:type_name -> uber.base.mergestrategy.Strategy - 11, // 2: uber.submitqueue.gateway.StatusResponse.metadata:type_name -> uber.submitqueue.gateway.StatusResponse.MetadataEntry - 8, // 3: uber.submitqueue.gateway.UnrecognizedQueueError.error:type_name -> uber.submitqueue.gateway.Error - 8, // 4: uber.submitqueue.gateway.RequestNotFoundError.error:type_name -> uber.submitqueue.gateway.Error - 0, // 5: uber.submitqueue.gateway.SubmitQueueGateway.Ping:input_type -> uber.submitqueue.gateway.PingRequest - 2, // 6: uber.submitqueue.gateway.SubmitQueueGateway.Land:input_type -> uber.submitqueue.gateway.LandRequest - 4, // 7: uber.submitqueue.gateway.SubmitQueueGateway.Cancel:input_type -> uber.submitqueue.gateway.CancelRequest - 6, // 8: uber.submitqueue.gateway.SubmitQueueGateway.Status:input_type -> uber.submitqueue.gateway.StatusRequest - 1, // 9: uber.submitqueue.gateway.SubmitQueueGateway.Ping:output_type -> uber.submitqueue.gateway.PingResponse - 3, // 10: uber.submitqueue.gateway.SubmitQueueGateway.Land:output_type -> uber.submitqueue.gateway.LandResponse - 5, // 11: uber.submitqueue.gateway.SubmitQueueGateway.Cancel:output_type -> uber.submitqueue.gateway.CancelResponse - 7, // 12: uber.submitqueue.gateway.SubmitQueueGateway.Status:output_type -> uber.submitqueue.gateway.StatusResponse - 9, // [9:13] is the sub-list for method output_type - 5, // [5:9] is the sub-list for method input_type - 5, // [5:5] is the sub-list for extension type_name - 5, // [5:5] is the sub-list for extension extendee - 0, // [0:5] is the sub-list for field type_name + 15, // 0: uber.submitqueue.gateway.LandRequest.change:type_name -> uber.base.change.Change + 16, // 1: uber.submitqueue.gateway.LandRequest.strategy:type_name -> uber.base.mergestrategy.Strategy + 14, // 2: uber.submitqueue.gateway.RequestSummary.metadata:type_name -> uber.submitqueue.gateway.RequestSummary.MetadataEntry + 6, // 3: uber.submitqueue.gateway.GetRequestSummaryByIDResponse.request:type_name -> uber.submitqueue.gateway.RequestSummary + 6, // 4: uber.submitqueue.gateway.GetRequestSummaryByChangeURIResponse.requests:type_name -> uber.submitqueue.gateway.RequestSummary + 11, // 5: uber.submitqueue.gateway.UnrecognizedQueueError.error:type_name -> uber.submitqueue.gateway.Error + 11, // 6: uber.submitqueue.gateway.RequestNotFoundError.error:type_name -> uber.submitqueue.gateway.Error + 0, // 7: uber.submitqueue.gateway.SubmitQueueGateway.Ping:input_type -> uber.submitqueue.gateway.PingRequest + 2, // 8: uber.submitqueue.gateway.SubmitQueueGateway.Land:input_type -> uber.submitqueue.gateway.LandRequest + 4, // 9: uber.submitqueue.gateway.SubmitQueueGateway.Cancel:input_type -> uber.submitqueue.gateway.CancelRequest + 7, // 10: uber.submitqueue.gateway.SubmitQueueGateway.GetRequestSummaryByID:input_type -> uber.submitqueue.gateway.GetRequestSummaryByIDRequest + 9, // 11: uber.submitqueue.gateway.SubmitQueueGateway.GetRequestSummaryByChangeURI:input_type -> uber.submitqueue.gateway.GetRequestSummaryByChangeURIRequest + 1, // 12: uber.submitqueue.gateway.SubmitQueueGateway.Ping:output_type -> uber.submitqueue.gateway.PingResponse + 3, // 13: uber.submitqueue.gateway.SubmitQueueGateway.Land:output_type -> uber.submitqueue.gateway.LandResponse + 5, // 14: uber.submitqueue.gateway.SubmitQueueGateway.Cancel:output_type -> uber.submitqueue.gateway.CancelResponse + 8, // 15: uber.submitqueue.gateway.SubmitQueueGateway.GetRequestSummaryByID:output_type -> uber.submitqueue.gateway.GetRequestSummaryByIDResponse + 10, // 16: uber.submitqueue.gateway.SubmitQueueGateway.GetRequestSummaryByChangeURI:output_type -> uber.submitqueue.gateway.GetRequestSummaryByChangeURIResponse + 12, // [12:17] is the sub-list for method output_type + 7, // [7:12] is the sub-list for method input_type + 7, // [7:7] is the sub-list for extension type_name + 7, // [7:7] is the sub-list for extension extendee + 0, // [0:7] is the sub-list for field type_name } func init() { file_gateway_proto_init() } @@ -743,7 +946,7 @@ func file_gateway_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_gateway_proto_rawDesc), len(file_gateway_proto_rawDesc)), NumEnums: 0, - NumMessages: 12, + NumMessages: 15, NumExtensions: 0, NumServices: 1, }, diff --git a/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go b/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go index a549dbd2..0eafe53b 100644 --- a/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go +++ b/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go @@ -24,7 +24,8 @@ type SubmitQueueGatewayYARPCClient interface { Ping(context.Context, *PingRequest, ...yarpc.CallOption) (*PingResponse, error) Land(context.Context, *LandRequest, ...yarpc.CallOption) (*LandResponse, error) Cancel(context.Context, *CancelRequest, ...yarpc.CallOption) (*CancelResponse, error) - Status(context.Context, *StatusRequest, ...yarpc.CallOption) (*StatusResponse, error) + GetRequestSummaryByID(context.Context, *GetRequestSummaryByIDRequest, ...yarpc.CallOption) (*GetRequestSummaryByIDResponse, error) + GetRequestSummaryByChangeURI(context.Context, *GetRequestSummaryByChangeURIRequest, ...yarpc.CallOption) (*GetRequestSummaryByChangeURIResponse, error) } func newSubmitQueueGatewayYARPCClient(clientConfig transport.ClientConfig, anyResolver v2.AnyResolver, options ...v2.ClientOption) SubmitQueueGatewayYARPCClient { @@ -48,7 +49,8 @@ type SubmitQueueGatewayYARPCServer interface { Ping(context.Context, *PingRequest) (*PingResponse, error) Land(context.Context, *LandRequest) (*LandResponse, error) Cancel(context.Context, *CancelRequest) (*CancelResponse, error) - Status(context.Context, *StatusRequest) (*StatusResponse, error) + GetRequestSummaryByID(context.Context, *GetRequestSummaryByIDRequest) (*GetRequestSummaryByIDResponse, error) + GetRequestSummaryByChangeURI(context.Context, *GetRequestSummaryByChangeURIRequest) (*GetRequestSummaryByChangeURIResponse, error) } type buildSubmitQueueGatewayYARPCProceduresParams struct { @@ -93,11 +95,21 @@ func buildSubmitQueueGatewayYARPCProcedures(params buildSubmitQueueGatewayYARPCP ), }, { - MethodName: "Status", + MethodName: "GetRequestSummaryByID", Handler: v2.NewUnaryHandler( v2.UnaryHandlerParams{ - Handle: handler.Status, - NewRequest: newSubmitQueueGatewayServiceStatusYARPCRequest, + Handle: handler.GetRequestSummaryByID, + NewRequest: newSubmitQueueGatewayServiceGetRequestSummaryByIDYARPCRequest, + AnyResolver: params.AnyResolver, + }, + ), + }, + { + MethodName: "GetRequestSummaryByChangeURI", + Handler: v2.NewUnaryHandler( + v2.UnaryHandlerParams{ + Handle: handler.GetRequestSummaryByChangeURI, + NewRequest: newSubmitQueueGatewayServiceGetRequestSummaryByChangeURIYARPCRequest, AnyResolver: params.AnyResolver, }, ), @@ -250,14 +262,26 @@ func (c *_SubmitQueueGatewayYARPCCaller) Cancel(ctx context.Context, request *Ca return response, err } -func (c *_SubmitQueueGatewayYARPCCaller) Status(ctx context.Context, request *StatusRequest, options ...yarpc.CallOption) (*StatusResponse, error) { - responseMessage, err := c.streamClient.Call(ctx, "Status", request, newSubmitQueueGatewayServiceStatusYARPCResponse, options...) +func (c *_SubmitQueueGatewayYARPCCaller) GetRequestSummaryByID(ctx context.Context, request *GetRequestSummaryByIDRequest, options ...yarpc.CallOption) (*GetRequestSummaryByIDResponse, error) { + responseMessage, err := c.streamClient.Call(ctx, "GetRequestSummaryByID", request, newSubmitQueueGatewayServiceGetRequestSummaryByIDYARPCResponse, options...) + if responseMessage == nil { + return nil, err + } + response, ok := responseMessage.(*GetRequestSummaryByIDResponse) + if !ok { + return nil, v2.CastError(emptySubmitQueueGatewayServiceGetRequestSummaryByIDYARPCResponse, responseMessage) + } + return response, err +} + +func (c *_SubmitQueueGatewayYARPCCaller) GetRequestSummaryByChangeURI(ctx context.Context, request *GetRequestSummaryByChangeURIRequest, options ...yarpc.CallOption) (*GetRequestSummaryByChangeURIResponse, error) { + responseMessage, err := c.streamClient.Call(ctx, "GetRequestSummaryByChangeURI", request, newSubmitQueueGatewayServiceGetRequestSummaryByChangeURIYARPCResponse, options...) if responseMessage == nil { return nil, err } - response, ok := responseMessage.(*StatusResponse) + response, ok := responseMessage.(*GetRequestSummaryByChangeURIResponse) if !ok { - return nil, v2.CastError(emptySubmitQueueGatewayServiceStatusYARPCResponse, responseMessage) + return nil, v2.CastError(emptySubmitQueueGatewayServiceGetRequestSummaryByChangeURIYARPCResponse, responseMessage) } return response, err } @@ -314,16 +338,32 @@ func (h *_SubmitQueueGatewayYARPCHandler) Cancel(ctx context.Context, requestMes return response, err } -func (h *_SubmitQueueGatewayYARPCHandler) Status(ctx context.Context, requestMessage proto.Message) (proto.Message, error) { - var request *StatusRequest +func (h *_SubmitQueueGatewayYARPCHandler) GetRequestSummaryByID(ctx context.Context, requestMessage proto.Message) (proto.Message, error) { + var request *GetRequestSummaryByIDRequest var ok bool if requestMessage != nil { - request, ok = requestMessage.(*StatusRequest) + request, ok = requestMessage.(*GetRequestSummaryByIDRequest) if !ok { - return nil, v2.CastError(emptySubmitQueueGatewayServiceStatusYARPCRequest, requestMessage) + return nil, v2.CastError(emptySubmitQueueGatewayServiceGetRequestSummaryByIDYARPCRequest, requestMessage) } } - response, err := h.server.Status(ctx, request) + response, err := h.server.GetRequestSummaryByID(ctx, request) + if response == nil { + return nil, err + } + return response, err +} + +func (h *_SubmitQueueGatewayYARPCHandler) GetRequestSummaryByChangeURI(ctx context.Context, requestMessage proto.Message) (proto.Message, error) { + var request *GetRequestSummaryByChangeURIRequest + var ok bool + if requestMessage != nil { + request, ok = requestMessage.(*GetRequestSummaryByChangeURIRequest) + if !ok { + return nil, v2.CastError(emptySubmitQueueGatewayServiceGetRequestSummaryByChangeURIYARPCRequest, requestMessage) + } + } + response, err := h.server.GetRequestSummaryByChangeURI(ctx, request) if response == nil { return nil, err } @@ -354,67 +394,88 @@ func newSubmitQueueGatewayServiceCancelYARPCResponse() proto.Message { return &CancelResponse{} } -func newSubmitQueueGatewayServiceStatusYARPCRequest() proto.Message { - return &StatusRequest{} +func newSubmitQueueGatewayServiceGetRequestSummaryByIDYARPCRequest() proto.Message { + return &GetRequestSummaryByIDRequest{} +} + +func newSubmitQueueGatewayServiceGetRequestSummaryByIDYARPCResponse() proto.Message { + return &GetRequestSummaryByIDResponse{} +} + +func newSubmitQueueGatewayServiceGetRequestSummaryByChangeURIYARPCRequest() proto.Message { + return &GetRequestSummaryByChangeURIRequest{} } -func newSubmitQueueGatewayServiceStatusYARPCResponse() proto.Message { - return &StatusResponse{} +func newSubmitQueueGatewayServiceGetRequestSummaryByChangeURIYARPCResponse() proto.Message { + return &GetRequestSummaryByChangeURIResponse{} } var ( - emptySubmitQueueGatewayServicePingYARPCRequest = &PingRequest{} - emptySubmitQueueGatewayServicePingYARPCResponse = &PingResponse{} - emptySubmitQueueGatewayServiceLandYARPCRequest = &LandRequest{} - emptySubmitQueueGatewayServiceLandYARPCResponse = &LandResponse{} - emptySubmitQueueGatewayServiceCancelYARPCRequest = &CancelRequest{} - emptySubmitQueueGatewayServiceCancelYARPCResponse = &CancelResponse{} - emptySubmitQueueGatewayServiceStatusYARPCRequest = &StatusRequest{} - emptySubmitQueueGatewayServiceStatusYARPCResponse = &StatusResponse{} + emptySubmitQueueGatewayServicePingYARPCRequest = &PingRequest{} + emptySubmitQueueGatewayServicePingYARPCResponse = &PingResponse{} + emptySubmitQueueGatewayServiceLandYARPCRequest = &LandRequest{} + emptySubmitQueueGatewayServiceLandYARPCResponse = &LandResponse{} + emptySubmitQueueGatewayServiceCancelYARPCRequest = &CancelRequest{} + emptySubmitQueueGatewayServiceCancelYARPCResponse = &CancelResponse{} + emptySubmitQueueGatewayServiceGetRequestSummaryByIDYARPCRequest = &GetRequestSummaryByIDRequest{} + emptySubmitQueueGatewayServiceGetRequestSummaryByIDYARPCResponse = &GetRequestSummaryByIDResponse{} + emptySubmitQueueGatewayServiceGetRequestSummaryByChangeURIYARPCRequest = &GetRequestSummaryByChangeURIRequest{} + emptySubmitQueueGatewayServiceGetRequestSummaryByChangeURIYARPCResponse = &GetRequestSummaryByChangeURIResponse{} ) var yarpcFileDescriptorClosuref1a937782ebbded5 = [][]byte{ // gateway.proto []byte{ - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xa4, 0x54, 0x5f, 0x6b, 0xd4, 0x40, - 0x10, 0x6f, 0x2e, 0xed, 0xd9, 0xce, 0xdd, 0x95, 0xb2, 0x94, 0x12, 0x42, 0xc5, 0x76, 0x45, 0x7b, - 0x4f, 0x49, 0x39, 0x51, 0xc4, 0xe2, 0x4b, 0x4b, 0xf5, 0x45, 0x4b, 0x4d, 0x11, 0x41, 0x90, 0xb2, - 0xc9, 0x0d, 0xb9, 0xd0, 0x4b, 0x72, 0x97, 0xdd, 0x54, 0xce, 0x67, 0xfd, 0x06, 0x7e, 0x38, 0x3f, - 0x8e, 0xec, 0x9f, 0xdc, 0x25, 0x60, 0xae, 0x0f, 0x3e, 0x65, 0x67, 0xf2, 0xfb, 0xcd, 0xcc, 0xce, - 0xfc, 0x66, 0x61, 0x10, 0x33, 0x81, 0xdf, 0xd9, 0xc2, 0x9b, 0x15, 0xb9, 0xc8, 0x89, 0x53, 0x86, - 0x58, 0x78, 0xbc, 0x0c, 0xd3, 0x44, 0xcc, 0x4b, 0x2c, 0xd1, 0x33, 0xff, 0x5d, 0xca, 0x66, 0x89, - 0x1f, 0x32, 0x8e, 0x7e, 0x34, 0x61, 0x59, 0x8c, 0xbe, 0x22, 0x18, 0x43, 0xb3, 0xdd, 0xd3, 0x25, - 0x26, 0xc5, 0x22, 0x46, 0x2e, 0x0a, 0x26, 0x30, 0x5e, 0x18, 0x68, 0xc3, 0xa7, 0x19, 0xf4, 0x04, - 0x7a, 0xd7, 0x49, 0x16, 0x07, 0x38, 0x2f, 0x91, 0x0b, 0xe2, 0xc0, 0xa3, 0x14, 0x39, 0x67, 0x31, - 0x3a, 0xd6, 0x91, 0x35, 0xdc, 0x09, 0x2a, 0x93, 0xfe, 0xb2, 0xa0, 0xaf, 0x91, 0x7c, 0x96, 0x67, - 0x1c, 0xdb, 0xa1, 0xe4, 0x18, 0xfa, 0x1c, 0x8b, 0xfb, 0x24, 0xc2, 0xdb, 0x8c, 0xa5, 0xe8, 0x74, - 0xd4, 0xef, 0x9e, 0xf1, 0x5d, 0xb1, 0x14, 0xc9, 0x21, 0xec, 0x88, 0x24, 0x45, 0x2e, 0x58, 0x3a, - 0x73, 0xec, 0x23, 0x6b, 0x68, 0x07, 0x2b, 0x07, 0x71, 0x61, 0x7b, 0x92, 0x73, 0xa1, 0xc8, 0x9b, - 0x8a, 0xbc, 0xb4, 0xe9, 0x6f, 0x0b, 0x7a, 0x1f, 0x58, 0x36, 0xae, 0x2a, 0xde, 0x87, 0x2d, 0xd5, - 0x27, 0x53, 0x84, 0x36, 0xc8, 0x29, 0x74, 0x75, 0x63, 0x54, 0xf2, 0xde, 0xc8, 0xf1, 0x54, 0x5f, - 0x65, 0x6b, 0x3c, 0xd3, 0xb1, 0x0b, 0xf5, 0x09, 0x0c, 0x8e, 0xbc, 0x85, 0xed, 0xaa, 0x35, 0x2a, - 0xe7, 0xee, 0xe8, 0xb8, 0xc6, 0x69, 0xb6, 0xee, 0xc6, 0x1c, 0x82, 0x25, 0x85, 0x52, 0xe8, 0xeb, - 0xaa, 0x4c, 0x77, 0x08, 0x6c, 0xf2, 0x79, 0x32, 0x36, 0x55, 0xa9, 0x33, 0x3d, 0x83, 0xc1, 0x05, - 0xcb, 0x22, 0x9c, 0x56, 0xb5, 0xff, 0x03, 0x44, 0x0e, 0xa0, 0x5b, 0x20, 0xe3, 0x79, 0x66, 0xda, - 0x66, 0x2c, 0xba, 0x07, 0xbb, 0x15, 0x59, 0xa7, 0xa0, 0x4f, 0x61, 0x70, 0x23, 0x98, 0x28, 0xf9, - 0x9a, 0x70, 0xf4, 0x8f, 0x05, 0xbb, 0x15, 0xca, 0x94, 0x76, 0x00, 0x5d, 0xae, 0x3c, 0x06, 0x68, - 0x2c, 0xf2, 0x18, 0x60, 0xca, 0xb8, 0xb8, 0xc5, 0xa2, 0xc8, 0x0b, 0x93, 0x7d, 0x47, 0x7a, 0x2e, - 0xa5, 0x83, 0x04, 0xb0, 0x9d, 0xa2, 0x60, 0x63, 0x26, 0x98, 0x63, 0x1f, 0xd9, 0xc3, 0xde, 0xe8, - 0x95, 0xd7, 0x26, 0x56, 0xaf, 0x99, 0xd2, 0xfb, 0x68, 0x88, 0x97, 0x99, 0x28, 0x16, 0xc1, 0x32, - 0x8e, 0x7b, 0x06, 0x83, 0xc6, 0x2f, 0xb2, 0x07, 0xf6, 0x1d, 0x2e, 0x4c, 0x61, 0xf2, 0x28, 0xe7, - 0x7b, 0xcf, 0xa6, 0x65, 0xa5, 0x22, 0x6d, 0xbc, 0xe9, 0xbc, 0xb6, 0xe8, 0x31, 0x6c, 0xe9, 0xca, - 0xda, 0x45, 0x8b, 0x70, 0xf0, 0x39, 0x2b, 0x30, 0xca, 0xe3, 0x2c, 0xf9, 0x81, 0xe3, 0x4f, 0xb2, - 0x46, 0xcd, 0x79, 0x09, 0x5b, 0xfa, 0x9e, 0x96, 0xd2, 0xc7, 0x93, 0xf6, 0xab, 0x28, 0x7c, 0xa0, - 0xd1, 0x2b, 0xb5, 0x75, 0x6a, 0x6a, 0xa3, 0x0c, 0xf6, 0xcd, 0x0c, 0xae, 0x72, 0xf1, 0x2e, 0x2f, - 0xb3, 0xf1, 0x7f, 0x25, 0xa9, 0xe6, 0xd8, 0x59, 0xcd, 0x71, 0xf4, 0xd3, 0x06, 0x72, 0xa3, 0x88, - 0xea, 0x12, 0xef, 0x35, 0x8f, 0x7c, 0x81, 0x4d, 0xb9, 0x94, 0xe4, 0x59, 0x7b, 0xe8, 0xda, 0x7a, - 0xbb, 0xcf, 0x1f, 0x82, 0x19, 0x69, 0x6d, 0xc8, 0xc0, 0x52, 0xcf, 0xeb, 0x02, 0xd7, 0xb6, 0x70, - 0x5d, 0xe0, 0xfa, 0x5a, 0xd0, 0x0d, 0xf2, 0x0d, 0xba, 0x5a, 0xc7, 0xe4, 0xa4, 0x9d, 0xd3, 0x58, - 0x13, 0x77, 0xf8, 0x30, 0xb0, 0x1e, 0x5e, 0x6b, 0x6f, 0x5d, 0xf8, 0xc6, 0xda, 0xac, 0x0b, 0xdf, - 0x94, 0x31, 0xdd, 0x38, 0xbf, 0x83, 0xc3, 0x28, 0x4f, 0x5b, 0x09, 0xe7, 0x7d, 0x33, 0x98, 0x6b, - 0xf9, 0xb8, 0x5e, 0x5b, 0x5f, 0xcf, 0xe2, 0x44, 0x4c, 0xca, 0xd0, 0x8b, 0xf2, 0xd4, 0x97, 0x24, - 0xbf, 0x46, 0xf2, 0xe5, 0x63, 0x5d, 0xb7, 0x4d, 0x10, 0xfd, 0x5a, 0xcf, 0xc2, 0xb0, 0xab, 0x0e, - 0x2f, 0xfe, 0x06, 0x00, 0x00, 0xff, 0xff, 0x01, 0x08, 0x76, 0xa7, 0x23, 0x06, 0x00, 0x00, + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xa4, 0x56, 0x5f, 0x4f, 0xfb, 0x36, + 0x14, 0x25, 0x4d, 0x5b, 0xda, 0xdb, 0x52, 0x21, 0x8b, 0xa1, 0x28, 0x02, 0x51, 0x3c, 0x36, 0xfa, + 0x94, 0xa2, 0x4e, 0xfb, 0xa3, 0x21, 0x26, 0xad, 0xc0, 0x10, 0xd2, 0x40, 0x2c, 0x08, 0x4d, 0x9a, + 0x34, 0x55, 0x6e, 0x6a, 0x85, 0x88, 0x26, 0x29, 0xb6, 0xc3, 0xd4, 0x3d, 0xed, 0x65, 0x2f, 0x7b, + 0xde, 0xfb, 0x3e, 0xcf, 0xde, 0xf7, 0x81, 0xa6, 0xd8, 0x4e, 0x48, 0x51, 0x53, 0xe0, 0xf7, 0x7b, + 0xaa, 0xaf, 0x7d, 0xcf, 0xb9, 0xc7, 0xc7, 0xf6, 0x6d, 0x60, 0xc3, 0x27, 0x82, 0xfe, 0x46, 0xe6, + 0xce, 0x8c, 0xc5, 0x22, 0x46, 0x56, 0x32, 0xa6, 0xcc, 0xe1, 0xc9, 0x38, 0x0c, 0xc4, 0x63, 0x42, + 0x13, 0xea, 0xe8, 0x75, 0x1b, 0x93, 0x59, 0xd0, 0x1f, 0x13, 0x4e, 0xfb, 0xde, 0x3d, 0x89, 0x7c, + 0xda, 0x97, 0x00, 0x1d, 0x28, 0xb4, 0x7d, 0x94, 0xe7, 0x84, 0x94, 0xf9, 0x94, 0x0b, 0x46, 0x04, + 0xf5, 0xe7, 0x3a, 0x75, 0x61, 0x4e, 0x21, 0xf0, 0x21, 0xb4, 0x6e, 0x82, 0xc8, 0x77, 0xe9, 0x63, + 0x42, 0xb9, 0x40, 0x16, 0xac, 0x87, 0x94, 0x73, 0xe2, 0x53, 0xcb, 0xe8, 0x1a, 0xbd, 0xa6, 0x9b, + 0x85, 0xf8, 0x4f, 0x03, 0xda, 0x2a, 0x93, 0xcf, 0xe2, 0x88, 0xd3, 0xf2, 0x54, 0xb4, 0x0f, 0x6d, + 0x4e, 0xd9, 0x53, 0xe0, 0xd1, 0x51, 0x44, 0x42, 0x6a, 0x55, 0xe4, 0x72, 0x4b, 0xcf, 0x5d, 0x93, + 0x90, 0xa2, 0x1d, 0x68, 0x8a, 0x20, 0xa4, 0x5c, 0x90, 0x70, 0x66, 0x99, 0x5d, 0xa3, 0x67, 0xba, + 0xcf, 0x13, 0xc8, 0x86, 0xc6, 0x7d, 0xcc, 0x85, 0x04, 0x57, 0x25, 0x38, 0x8f, 0xf1, 0xdf, 0x06, + 0xb4, 0x7e, 0x24, 0xd1, 0x24, 0x53, 0xbc, 0x05, 0x35, 0xe9, 0x93, 0x16, 0xa1, 0x02, 0x74, 0x04, + 0x75, 0x65, 0x8c, 0x2c, 0xde, 0x1a, 0x58, 0x8e, 0xf4, 0x35, 0xb5, 0xc6, 0xd1, 0x8e, 0x9d, 0xca, + 0x1f, 0x57, 0xe7, 0xa1, 0x13, 0x68, 0x64, 0xd6, 0xc8, 0x9a, 0x9d, 0xc1, 0x7e, 0x01, 0xb3, 0x68, + 0xdd, 0xad, 0x1e, 0xb8, 0x39, 0x04, 0x63, 0x68, 0x2b, 0x55, 0xda, 0x1d, 0x04, 0x55, 0xfe, 0x18, + 0x4c, 0xb4, 0x2a, 0x39, 0xc6, 0xc7, 0xb0, 0x71, 0x4a, 0x22, 0x8f, 0x4e, 0x33, 0xed, 0x4b, 0x92, + 0xd0, 0x36, 0xd4, 0x19, 0x25, 0x3c, 0x8e, 0xb4, 0x6d, 0x3a, 0xc2, 0x9b, 0xd0, 0xc9, 0xc0, 0xaa, + 0x04, 0xfe, 0xb7, 0x02, 0x1d, 0xcd, 0x74, 0x9b, 0x84, 0x21, 0x61, 0xf3, 0xa5, 0x84, 0xb9, 0x41, + 0x95, 0xa2, 0x41, 0x7b, 0xd0, 0x52, 0x1b, 0x1f, 0x25, 0x2c, 0xe0, 0x96, 0xd9, 0x35, 0x7b, 0x4d, + 0x17, 0xd4, 0xd4, 0x1d, 0x0b, 0x38, 0x3a, 0x80, 0x0e, 0xa3, 0x1e, 0x0d, 0x9e, 0xe8, 0x64, 0x44, + 0xc4, 0x28, 0xe4, 0xd2, 0x15, 0xd3, 0x6d, 0x67, 0xb3, 0xdf, 0x8b, 0x2b, 0x9e, 0xaa, 0xe5, 0x82, + 0x88, 0x84, 0x5b, 0x35, 0xa5, 0x56, 0x45, 0x68, 0x17, 0x60, 0x4a, 0xb8, 0x18, 0x51, 0xc6, 0x62, + 0x66, 0xd5, 0xe5, 0x5a, 0x33, 0x9d, 0x39, 0x4f, 0x27, 0x90, 0x0b, 0x8d, 0x90, 0x0a, 0x32, 0x21, + 0x82, 0x58, 0xeb, 0x5d, 0xb3, 0xd7, 0x1a, 0x7c, 0xe5, 0x94, 0x5d, 0x7c, 0x67, 0x71, 0x8f, 0xce, + 0x95, 0x06, 0x9e, 0x47, 0x82, 0xcd, 0xdd, 0x9c, 0xc7, 0x3e, 0x86, 0x8d, 0x85, 0x25, 0xb4, 0x09, + 0xe6, 0x03, 0x9d, 0x6b, 0x2f, 0xd2, 0x61, 0x6a, 0xc5, 0x13, 0x99, 0x3e, 0x5b, 0x21, 0x83, 0x6f, + 0x2b, 0xdf, 0x18, 0x78, 0x00, 0x3b, 0x17, 0x54, 0x2c, 0x56, 0x1a, 0xce, 0x2f, 0xcf, 0x56, 0x9c, + 0x14, 0xf6, 0x60, 0xb7, 0x04, 0xa3, 0xef, 0xc0, 0x10, 0xd6, 0x99, 0x5a, 0x95, 0xb8, 0xd6, 0xa0, + 0xf7, 0xd6, 0x4d, 0xba, 0x19, 0x10, 0x9f, 0xc1, 0xa7, 0x4b, 0x8a, 0xa8, 0xbb, 0x7b, 0xe7, 0x5e, + 0x66, 0xfa, 0x76, 0x01, 0x9e, 0x8f, 0x53, 0xab, 0x6c, 0xe6, 0xa7, 0x89, 0xa7, 0x70, 0xb0, 0x9a, + 0x45, 0x2b, 0x3e, 0x83, 0x86, 0x2e, 0xcc, 0x2d, 0x43, 0x9e, 0xcb, 0xdb, 0x25, 0xe7, 0x48, 0xbc, + 0x0f, 0x35, 0x75, 0xcc, 0xe5, 0xdd, 0x84, 0xc2, 0xf6, 0x5d, 0xc4, 0xa8, 0x17, 0xfb, 0x51, 0xf0, + 0x3b, 0x9d, 0xfc, 0x94, 0x12, 0x2b, 0xcc, 0x97, 0x50, 0x53, 0x97, 0x46, 0x59, 0xb6, 0x57, 0x5e, + 0x5f, 0xe6, 0xbb, 0x2a, 0x7b, 0xf9, 0x2d, 0xc7, 0x7f, 0x18, 0xb0, 0xa5, 0x65, 0x5e, 0xc7, 0xe2, + 0x87, 0x38, 0x89, 0x26, 0x1f, 0x55, 0x25, 0xbb, 0x06, 0x95, 0xc2, 0xfb, 0x5a, 0xb4, 0xde, 0x7c, + 0x61, 0xfd, 0xe0, 0xbf, 0x2a, 0xa0, 0x5b, 0xc9, 0x2b, 0x37, 0x79, 0xa1, 0x68, 0xd1, 0xcf, 0x50, + 0x4d, 0xbb, 0x29, 0xfa, 0xac, 0xbc, 0x72, 0xa1, 0x2f, 0xdb, 0x9f, 0xbf, 0x96, 0xa6, 0x7b, 0xc2, + 0x5a, 0x4a, 0x9c, 0x36, 0xa2, 0x55, 0xc4, 0x85, 0xf6, 0xb9, 0x8a, 0xb8, 0xd8, 0xcf, 0xf0, 0x1a, + 0xfa, 0x15, 0xea, 0xaa, 0x01, 0xa1, 0xc3, 0x72, 0xcc, 0x42, 0x7f, 0xb3, 0x7b, 0xaf, 0x27, 0xe6, + 0xf4, 0x7f, 0x19, 0xf0, 0xc9, 0xd2, 0xe7, 0x84, 0x56, 0xb4, 0x86, 0x55, 0x6f, 0xd6, 0xfe, 0xfa, + 0xdd, 0xb8, 0x5c, 0xcc, 0x3f, 0xc6, 0xd2, 0x7e, 0x90, 0x3f, 0x18, 0x74, 0xf2, 0x2e, 0xee, 0x97, + 0xcf, 0xd5, 0xfe, 0xee, 0x43, 0xe1, 0x99, 0xc2, 0xe1, 0x03, 0xec, 0x78, 0x71, 0x58, 0x4a, 0x33, + 0x6c, 0xeb, 0x8b, 0x76, 0x93, 0xfe, 0xcb, 0xdf, 0x18, 0xbf, 0x1c, 0xfb, 0x81, 0xb8, 0x4f, 0xc6, + 0x8e, 0x17, 0x87, 0xfd, 0x14, 0xd4, 0x2f, 0x80, 0xfa, 0xe9, 0x57, 0x43, 0x31, 0xd6, 0x24, 0xea, + 0xb3, 0x61, 0x36, 0x1e, 0xd7, 0xe5, 0xe0, 0x8b, 0xff, 0x03, 0x00, 0x00, 0xff, 0xff, 0x40, 0x17, + 0x90, 0xe2, 0xac, 0x08, 0x00, 0x00, }, // api/base/change/proto/change.proto []byte{ diff --git a/api/submitqueue/gateway/protopb/gateway_grpc.pb.go b/api/submitqueue/gateway/protopb/gateway_grpc.pb.go index f271309f..8cd12a7f 100644 --- a/api/submitqueue/gateway/protopb/gateway_grpc.pb.go +++ b/api/submitqueue/gateway/protopb/gateway_grpc.pb.go @@ -34,10 +34,11 @@ import ( const _ = grpc.SupportPackageIsVersion9 const ( - SubmitQueueGateway_Ping_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/Ping" - SubmitQueueGateway_Land_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/Land" - SubmitQueueGateway_Cancel_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/Cancel" - SubmitQueueGateway_Status_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/Status" + SubmitQueueGateway_Ping_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/Ping" + SubmitQueueGateway_Land_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/Land" + SubmitQueueGateway_Cancel_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/Cancel" + SubmitQueueGateway_GetRequestSummaryByID_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/GetRequestSummaryByID" + SubmitQueueGateway_GetRequestSummaryByChangeURI_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/GetRequestSummaryByChangeURI" ) // SubmitQueueGatewayClient is the client API for SubmitQueueGateway service. @@ -61,11 +62,12 @@ type SubmitQueueGatewayClient interface { // Cancellation is NOT GUARANTEED: a request that has already merged, or that races to completion before the cancel // signal propagates through the pipeline, may still land (or end in an error). Callers must NOT assume that a // successful Cancel response means the request was cancelled — the actual terminal outcome (cancelled, landed, or - // error) must be checked through the separate status / request-log API. + // error) must be checked through the request-summary or request-history APIs. Cancel(ctx context.Context, in *CancelRequest, opts ...grpc.CallOption) (*CancelResponse, error) - // Status returns the current status of a previously submitted request, identified by its sqid. - // The status is eventually consistent with the request store and reconciled from the append-only request log. - Status(ctx context.Context, in *StatusRequest, opts ...grpc.CallOption) (*StatusResponse, error) + // GetRequestSummaryByID returns the current materialized status of one request. + GetRequestSummaryByID(ctx context.Context, in *GetRequestSummaryByIDRequest, opts ...grpc.CallOption) (*GetRequestSummaryByIDResponse, error) + // GetRequestSummaryByChangeURI returns current materialized statuses for an exact pinned change URI. + GetRequestSummaryByChangeURI(ctx context.Context, in *GetRequestSummaryByChangeURIRequest, opts ...grpc.CallOption) (*GetRequestSummaryByChangeURIResponse, error) } type submitQueueGatewayClient struct { @@ -106,10 +108,20 @@ func (c *submitQueueGatewayClient) Cancel(ctx context.Context, in *CancelRequest return out, nil } -func (c *submitQueueGatewayClient) Status(ctx context.Context, in *StatusRequest, opts ...grpc.CallOption) (*StatusResponse, error) { +func (c *submitQueueGatewayClient) GetRequestSummaryByID(ctx context.Context, in *GetRequestSummaryByIDRequest, opts ...grpc.CallOption) (*GetRequestSummaryByIDResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) - out := new(StatusResponse) - err := c.cc.Invoke(ctx, SubmitQueueGateway_Status_FullMethodName, in, out, cOpts...) + out := new(GetRequestSummaryByIDResponse) + err := c.cc.Invoke(ctx, SubmitQueueGateway_GetRequestSummaryByID_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *submitQueueGatewayClient) GetRequestSummaryByChangeURI(ctx context.Context, in *GetRequestSummaryByChangeURIRequest, opts ...grpc.CallOption) (*GetRequestSummaryByChangeURIResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(GetRequestSummaryByChangeURIResponse) + err := c.cc.Invoke(ctx, SubmitQueueGateway_GetRequestSummaryByChangeURI_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } @@ -137,11 +149,12 @@ type SubmitQueueGatewayServer interface { // Cancellation is NOT GUARANTEED: a request that has already merged, or that races to completion before the cancel // signal propagates through the pipeline, may still land (or end in an error). Callers must NOT assume that a // successful Cancel response means the request was cancelled — the actual terminal outcome (cancelled, landed, or - // error) must be checked through the separate status / request-log API. + // error) must be checked through the request-summary or request-history APIs. Cancel(context.Context, *CancelRequest) (*CancelResponse, error) - // Status returns the current status of a previously submitted request, identified by its sqid. - // The status is eventually consistent with the request store and reconciled from the append-only request log. - Status(context.Context, *StatusRequest) (*StatusResponse, error) + // GetRequestSummaryByID returns the current materialized status of one request. + GetRequestSummaryByID(context.Context, *GetRequestSummaryByIDRequest) (*GetRequestSummaryByIDResponse, error) + // GetRequestSummaryByChangeURI returns current materialized statuses for an exact pinned change URI. + GetRequestSummaryByChangeURI(context.Context, *GetRequestSummaryByChangeURIRequest) (*GetRequestSummaryByChangeURIResponse, error) mustEmbedUnimplementedSubmitQueueGatewayServer() } @@ -161,8 +174,11 @@ func (UnimplementedSubmitQueueGatewayServer) Land(context.Context, *LandRequest) func (UnimplementedSubmitQueueGatewayServer) Cancel(context.Context, *CancelRequest) (*CancelResponse, error) { return nil, status.Errorf(codes.Unimplemented, "method Cancel not implemented") } -func (UnimplementedSubmitQueueGatewayServer) Status(context.Context, *StatusRequest) (*StatusResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method Status not implemented") +func (UnimplementedSubmitQueueGatewayServer) GetRequestSummaryByID(context.Context, *GetRequestSummaryByIDRequest) (*GetRequestSummaryByIDResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method GetRequestSummaryByID not implemented") +} +func (UnimplementedSubmitQueueGatewayServer) GetRequestSummaryByChangeURI(context.Context, *GetRequestSummaryByChangeURIRequest) (*GetRequestSummaryByChangeURIResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method GetRequestSummaryByChangeURI not implemented") } func (UnimplementedSubmitQueueGatewayServer) mustEmbedUnimplementedSubmitQueueGatewayServer() {} func (UnimplementedSubmitQueueGatewayServer) testEmbeddedByValue() {} @@ -239,20 +255,38 @@ func _SubmitQueueGateway_Cancel_Handler(srv interface{}, ctx context.Context, de return interceptor(ctx, in, info, handler) } -func _SubmitQueueGateway_Status_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { - in := new(StatusRequest) +func _SubmitQueueGateway_GetRequestSummaryByID_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(GetRequestSummaryByIDRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { - return srv.(SubmitQueueGatewayServer).Status(ctx, in) + return srv.(SubmitQueueGatewayServer).GetRequestSummaryByID(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, - FullMethod: SubmitQueueGateway_Status_FullMethodName, + FullMethod: SubmitQueueGateway_GetRequestSummaryByID_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { - return srv.(SubmitQueueGatewayServer).Status(ctx, req.(*StatusRequest)) + return srv.(SubmitQueueGatewayServer).GetRequestSummaryByID(ctx, req.(*GetRequestSummaryByIDRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _SubmitQueueGateway_GetRequestSummaryByChangeURI_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(GetRequestSummaryByChangeURIRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(SubmitQueueGatewayServer).GetRequestSummaryByChangeURI(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: SubmitQueueGateway_GetRequestSummaryByChangeURI_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(SubmitQueueGatewayServer).GetRequestSummaryByChangeURI(ctx, req.(*GetRequestSummaryByChangeURIRequest)) } return interceptor(ctx, in, info, handler) } @@ -277,8 +311,12 @@ var SubmitQueueGateway_ServiceDesc = grpc.ServiceDesc{ Handler: _SubmitQueueGateway_Cancel_Handler, }, { - MethodName: "Status", - Handler: _SubmitQueueGateway_Status_Handler, + MethodName: "GetRequestSummaryByID", + Handler: _SubmitQueueGateway_GetRequestSummaryByID_Handler, + }, + { + MethodName: "GetRequestSummaryByChangeURI", + Handler: _SubmitQueueGateway_GetRequestSummaryByChangeURI_Handler, }, }, Streams: []grpc.StreamDesc{}, diff --git a/service/submitqueue/gateway/server/BUILD.bazel b/service/submitqueue/gateway/server/BUILD.bazel index 9d24713c..39241b0a 100644 --- a/service/submitqueue/gateway/server/BUILD.bazel +++ b/service/submitqueue/gateway/server/BUILD.bazel @@ -1,4 +1,4 @@ -load("@rules_go//go:def.bzl", "go_binary", "go_library") +load("@rules_go//go:def.bzl", "go_binary", "go_library", "go_test") exports_files( ["docker-compose.yml"], @@ -40,3 +40,15 @@ go_binary( embed = [":gateway_lib"], visibility = ["//visibility:public"], ) + +go_test( + name = "go_default_test", + srcs = ["main_test.go"], + embed = [":gateway_lib"], # keep + deps = [ + "//submitqueue/gateway/controller:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@org_golang_google_grpc//codes:go_default_library", + "@org_golang_google_grpc//status:go_default_library", + ], +) diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index 427c76dd..45f9c638 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -52,10 +52,10 @@ 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 - statusController *controller.StatusController + pingController *controller.PingController + landController *controller.LandController + cancelController *controller.CancelController + requestSummaryController *controller.RequestSummaryController } // Ping delegates to the controller @@ -86,14 +86,37 @@ func (s *GatewayServer) Cancel(ctx context.Context, req *pb.CancelRequest) (*pb. return &pb.CancelResponse{}, nil } -// Status maps the wire request to an entity, delegates to the controller, and -// maps the read-model result back to the wire response. -func (s *GatewayServer) Status(ctx context.Context, req *pb.StatusRequest) (*pb.StatusResponse, error) { - state, err := s.statusController.Status(ctx, mapper.ProtoToStatusRequest(req)) +// GetRequestSummaryByID maps the wire request to an entity, delegates to the controller, and maps the result back to the wire response. +func (s *GatewayServer) GetRequestSummaryByID(ctx context.Context, req *pb.GetRequestSummaryByIDRequest) (*pb.GetRequestSummaryByIDResponse, error) { + summary, err := s.requestSummaryController.GetRequestSummaryByID(ctx, mapper.ProtoToGetRequestSummaryByIDRequest(req)) if err != nil { return nil, err } - return mapper.CurrentStateToProto(state), nil + return &pb.GetRequestSummaryByIDResponse{Request: mapper.RequestSummaryToProto(summary)}, nil +} + +// GetRequestSummaryByChangeURI maps the wire request to an entity, delegates to the controller, and maps the results back to the wire response. +func (s *GatewayServer) GetRequestSummaryByChangeURI(ctx context.Context, req *pb.GetRequestSummaryByChangeURIRequest) (*pb.GetRequestSummaryByChangeURIResponse, error) { + summaries, err := s.requestSummaryController.GetRequestSummaryByChangeURI(ctx, mapper.ProtoToGetRequestSummaryByChangeURIRequest(req)) + if err != nil { + return nil, err + } + return &pb.GetRequestSummaryByChangeURIResponse{Requests: mapper.RequestSummariesToProto(summaries)}, nil +} + +func gatewayStatusError(err error) error { + switch { + case controller.IsRequestNotFound(err): + return status.Error(codes.NotFound, err.Error()) + case controller.IsTooManyChangeRequests(err): + return status.Error(codes.ResourceExhausted, err.Error()) + case controller.IsInternalConsistency(err): + return status.Error(codes.Internal, err.Error()) + case controller.IsInvalidRequest(err), controller.IsUnrecognizedQueue(err): + return status.Error(codes.InvalidArgument, err.Error()) + default: + return err + } } func main() { @@ -234,29 +257,26 @@ func run() error { return fmt.Errorf("failed to create topic registry: %w", err) } - // Create gRPC server with a unary interceptor that translates user-input - // validation errors (anything in the chain that matches controller.ErrInvalidRequest) - // into codes.InvalidArgument so gRPC clients can distinguish bad input from - // infrastructure failures. Other errors pass through unchanged. + // Create gRPC server with a unary interceptor that translates gateway + // controller errors into stable transport status codes. grpcServer := grpc.NewServer(grpc.UnaryInterceptor( func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { resp, err := handler(ctx, req) - if err != nil && controller.IsInvalidRequest(err) { - return nil, status.Error(codes.InvalidArgument, err.Error()) + if err != nil { + return nil, gatewayStatusError(err) } - return resp, err + return resp, nil }, )) // Initialize storage from the shared app database connection. The land - // controller writes to this store directly; cancel/status use the request - // log store directly. The log consumer registered below persists request - // log entries published by the orchestrator's normal pipeline. + // controller writes to this store directly; cancel and request-summary + // controllers use its request stores. The log consumer registered below + // persists entries published by the orchestrator's normal pipeline. store, err := mysqlstorage.NewStorage(appDB, scope.SubScope("storage")) if err != nil { return fmt.Errorf("failed to create storage: %w", err) } - requestLogStore := store.GetRequestLogStore() // Load queue configurations from YAML. Path is required so the gateway // can reject requests for unknown queues at the edge. @@ -273,12 +293,17 @@ func run() error { pingController := controller.NewPingController(logger, scope) landController := controller.NewLandController(logger.Sugar(), scope, cnt, store, queueConfigs, registry) cancelController := controller.NewCancelController(logger.Sugar(), scope, store, registry) - statusController := controller.NewStatusController(logger.Sugar(), scope, requestLogStore) + requestSummaryController := controller.NewRequestSummaryController( + logger.Sugar(), + scope, + store.GetRequestSummaryStore(), + store.GetRequestURIStore(), + ) gatewayServer := &GatewayServer{ - pingController: pingController, - landController: landController, - cancelController: cancelController, - statusController: statusController, + pingController: pingController, + landController: landController, + cancelController: cancelController, + requestSummaryController: requestSummaryController, } pb.RegisterSubmitQueueGatewayServer(grpcServer, gatewayServer) diff --git a/service/submitqueue/gateway/server/main_test.go b/service/submitqueue/gateway/server/main_test.go new file mode 100644 index 00000000..1f96500a --- /dev/null +++ b/service/submitqueue/gateway/server/main_test.go @@ -0,0 +1,68 @@ +// 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 main + +import ( + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/uber/submitqueue/submitqueue/gateway/controller" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func TestGatewayStatusError(t *testing.T) { + tests := []struct { + name string + err error + code codes.Code + }{ + { + name: "invalid request", + err: controller.ErrInvalidRequest, + code: codes.InvalidArgument, + }, + { + name: "unrecognized queue", + err: &controller.UnrecognizedQueueError{Queue: "missing"}, + code: codes.InvalidArgument, + }, + { + name: "request not found", + err: &controller.RequestNotFoundError{Sqid: "queue/1"}, + code: codes.NotFound, + }, + { + name: "too many change requests", + err: &controller.TooManyChangeRequestsError{ChangeURI: "uri", Limit: 100}, + code: codes.ResourceExhausted, + }, + { + name: "internal consistency", + err: &controller.InternalConsistencyError{Message: "inconsistent"}, + code: codes.Internal, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.code, status.Code(gatewayStatusError(tt.err))) + }) + } + + infraErr := errors.New("storage unavailable") + assert.Equal(t, infraErr, gatewayStatusError(infraErr)) +} diff --git a/service/submitqueue/gateway/server/mapper/BUILD.bazel b/service/submitqueue/gateway/server/mapper/BUILD.bazel index a8f79d5c..572a438a 100644 --- a/service/submitqueue/gateway/server/mapper/BUILD.bazel +++ b/service/submitqueue/gateway/server/mapper/BUILD.bazel @@ -5,7 +5,7 @@ go_library( srcs = [ "cancel.go", "land.go", - "status.go", + "request_summary.go", ], importpath = "github.com/uber/submitqueue/service/submitqueue/gateway/server/mapper", visibility = ["//visibility:public"], @@ -14,7 +14,6 @@ go_library( "//api/submitqueue/gateway/protopb:go_default_library", "//platform/base/change:go_default_library", "//platform/base/mergestrategy:go_default_library", - "//submitqueue/core/request:go_default_library", "//submitqueue/entity:go_default_library", ], ) @@ -24,7 +23,7 @@ go_test( srcs = [ "cancel_test.go", "land_test.go", - "status_test.go", + "request_summary_test.go", ], embed = [":go_default_library"], deps = [ @@ -33,7 +32,6 @@ go_test( "//api/submitqueue/gateway/protopb:go_default_library", "//platform/base/change:go_default_library", "//platform/base/mergestrategy:go_default_library", - "//submitqueue/core/request:go_default_library", "//submitqueue/entity:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", diff --git a/service/submitqueue/gateway/server/mapper/land.go b/service/submitqueue/gateway/server/mapper/land.go index ea3fc0d6..63264956 100644 --- a/service/submitqueue/gateway/server/mapper/land.go +++ b/service/submitqueue/gateway/server/mapper/land.go @@ -14,7 +14,7 @@ // Package mapper translates gateway wire (proto) types to and from the domain // entities the controllers operate on. Each RPC gets its own file (land.go, -// status.go, cancel.go, …); translation lives here so controllers stay +// request_summary.go and cancel.go); translation lives here so controllers stay // proto-free. package mapper diff --git a/service/submitqueue/gateway/server/mapper/request_summary.go b/service/submitqueue/gateway/server/mapper/request_summary.go new file mode 100644 index 00000000..55204ade --- /dev/null +++ b/service/submitqueue/gateway/server/mapper/request_summary.go @@ -0,0 +1,67 @@ +// 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 ( + pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/submitqueue/entity" +) + +// ProtoToGetRequestSummaryByIDRequest maps the wire request to the entity request the controller operates on. +func ProtoToGetRequestSummaryByIDRequest(req *pb.GetRequestSummaryByIDRequest) entity.GetRequestSummaryByIDRequest { + return entity.GetRequestSummaryByIDRequest{ + ID: req.GetSqid(), + } +} + +// ProtoToGetRequestSummaryByChangeURIRequest maps the wire request to the entity request the controller operates on. +func ProtoToGetRequestSummaryByChangeURIRequest(req *pb.GetRequestSummaryByChangeURIRequest) entity.GetRequestSummaryByChangeURIRequest { + return entity.GetRequestSummaryByChangeURIRequest{ + ChangeURI: req.GetChangeUri(), + } +} + +// RequestSummaryToProto maps a domain request summary to the wire request summary. +func RequestSummaryToProto(summary entity.RequestSummary) *pb.RequestSummary { + return &pb.RequestSummary{ + Sqid: summary.RequestID, + Queue: summary.Queue, + ChangeUris: append([]string{}, summary.ChangeURIs...), + ReceivedAtMs: summary.ReceivedAtMs, + Status: string(summary.Status), + LastError: summary.LastError, + Metadata: cloneStringMap(summary.Metadata), + } +} + +// RequestSummariesToProto maps domain request summaries to wire request summaries. +func RequestSummariesToProto(summaries []entity.RequestSummary) []*pb.RequestSummary { + requests := make([]*pb.RequestSummary, 0, len(summaries)) + for _, summary := range summaries { + requests = append(requests, RequestSummaryToProto(summary)) + } + return requests +} + +func cloneStringMap(input map[string]string) map[string]string { + if input == nil { + return map[string]string{} + } + output := make(map[string]string, len(input)) + for key, value := range input { + output[key] = value + } + return output +} diff --git a/service/submitqueue/gateway/server/mapper/request_summary_test.go b/service/submitqueue/gateway/server/mapper/request_summary_test.go new file mode 100644 index 00000000..ccb600c2 --- /dev/null +++ b/service/submitqueue/gateway/server/mapper/request_summary_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 mapper + +import ( + "testing" + + "github.com/stretchr/testify/assert" + pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/submitqueue/entity" +) + +func TestProtoToGetRequestSummaryByIDRequest(t *testing.T) { + assert.Equal(t, + entity.GetRequestSummaryByIDRequest{ID: "test-queue/42"}, + ProtoToGetRequestSummaryByIDRequest(&pb.GetRequestSummaryByIDRequest{Sqid: "test-queue/42"}), + ) +} + +func TestProtoToGetRequestSummaryByChangeURIRequest(t *testing.T) { + assert.Equal(t, + entity.GetRequestSummaryByChangeURIRequest{ChangeURI: "github://uber/repo/pull/1/abc"}, + ProtoToGetRequestSummaryByChangeURIRequest(&pb.GetRequestSummaryByChangeURIRequest{ChangeUri: "github://uber/repo/pull/1/abc"}), + ) +} + +func TestRequestSummaryToProto(t *testing.T) { + summary := entity.RequestSummary{ + RequestID: "test-queue/42", + Queue: "test-queue", + ChangeURIs: []string{"github://uber/repo/pull/1/abc"}, + ReceivedAtMs: 100, + Status: entity.RequestStatusValidating, + LastError: "validation failed", + Metadata: map[string]string{"step": "lint"}, + } + + assert.Equal(t, &pb.RequestSummary{ + Sqid: "test-queue/42", + Queue: "test-queue", + ChangeUris: []string{"github://uber/repo/pull/1/abc"}, + ReceivedAtMs: 100, + Status: string(entity.RequestStatusValidating), + LastError: "validation failed", + Metadata: map[string]string{"step": "lint"}, + }, RequestSummaryToProto(summary)) +} + +func TestRequestSummariesToProto(t *testing.T) { + summaries := []entity.RequestSummary{ + {RequestID: "test-queue/2", Status: entity.RequestStatusLanded}, + {RequestID: "test-queue/1", Status: entity.RequestStatusError}, + } + + requests := RequestSummariesToProto(summaries) + + assert.Equal(t, []string{"test-queue/2", "test-queue/1"}, []string{requests[0].Sqid, requests[1].Sqid}) +} diff --git a/service/submitqueue/gateway/server/mapper/status.go b/service/submitqueue/gateway/server/mapper/status.go deleted file mode 100644 index d0c92593..00000000 --- a/service/submitqueue/gateway/server/mapper/status.go +++ /dev/null @@ -1,38 +0,0 @@ -// 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 ( - pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" - "github.com/uber/submitqueue/submitqueue/core/request" - "github.com/uber/submitqueue/submitqueue/entity" -) - -// ProtoToStatusRequest maps the wire StatusRequest to the entity.StatusRequest -// the controller operates on. -func ProtoToStatusRequest(req *pb.StatusRequest) entity.StatusRequest { - return entity.StatusRequest{ - ID: req.GetSqid(), - } -} - -// CurrentStateToProto maps the domain read model to the wire StatusResponse. -func CurrentStateToProto(state request.CurrentState) *pb.StatusResponse { - return &pb.StatusResponse{ - Status: string(state.Status), - LastError: state.LastError, - Metadata: state.Metadata, - } -} diff --git a/service/submitqueue/gateway/server/mapper/status_test.go b/service/submitqueue/gateway/server/mapper/status_test.go deleted file mode 100644 index 6c25aeef..00000000 --- a/service/submitqueue/gateway/server/mapper/status_test.go +++ /dev/null @@ -1,86 +0,0 @@ -// 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/submitqueue/gateway/protopb" - "github.com/uber/submitqueue/submitqueue/core/request" - "github.com/uber/submitqueue/submitqueue/entity" -) - -func TestProtoToStatusRequest(t *testing.T) { - tests := []struct { - name string - req *pb.StatusRequest - expected entity.StatusRequest - }{ - { - name: "maps sqid to ID", - req: &pb.StatusRequest{Sqid: "test-queue/42"}, - expected: entity.StatusRequest{ID: "test-queue/42"}, - }, - { - name: "empty request yields zero value", - req: &pb.StatusRequest{}, - expected: entity.StatusRequest{}, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - got := ProtoToStatusRequest(tt.req) - assert.Equal(t, tt.expected, got) - }) - } -} - -func TestCurrentStateToProto(t *testing.T) { - tests := []struct { - name string - state request.CurrentState - expected *pb.StatusResponse - }{ - { - name: "maps all fields", - state: request.CurrentState{ - Status: entity.RequestStatusValidating, - LastError: "validation failed", - Metadata: map[string]string{"step": "lint"}, - }, - expected: &pb.StatusResponse{ - Status: string(entity.RequestStatusValidating), - LastError: "validation failed", - Metadata: map[string]string{"step": "lint"}, - }, - }, - { - name: "zero value state maps to empty response", - state: request.CurrentState{}, - expected: &pb.StatusResponse{ - Status: "", - }, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - got := CurrentStateToProto(tt.state) - assert.Equal(t, tt.expected, got) - }) - } -} diff --git a/submitqueue/entity/BUILD.bazel b/submitqueue/entity/BUILD.bazel index 2bd60c84..030209da 100644 --- a/submitqueue/entity/BUILD.bazel +++ b/submitqueue/entity/BUILD.bazel @@ -21,7 +21,6 @@ go_library( "request_summary.go", "speculation_path_build.go", "speculation_tree.go", - "status.go", ], importpath = "github.com/uber/submitqueue/submitqueue/entity", visibility = ["//visibility:public"], diff --git a/submitqueue/entity/request_summary.go b/submitqueue/entity/request_summary.go index fe9565c0..e8a75009 100644 --- a/submitqueue/entity/request_summary.go +++ b/submitqueue/entity/request_summary.go @@ -14,6 +14,18 @@ package entity +// GetRequestSummaryByIDRequest identifies one request summary by sqid. +type GetRequestSummaryByIDRequest struct { + // ID is the globally unique identifier of the request. Format: "/". + ID string +} + +// GetRequestSummaryByChangeURIRequest identifies request summaries by an exact pinned change URI. +type GetRequestSummaryByChangeURIRequest struct { + // ChangeURI is the exact change URI supplied in a Land request. + ChangeURI string +} + // RequestSummary is the gateway-owned materialized current view of a request. // RequestID is exposed as sqid by the gateway API. type RequestSummary struct { diff --git a/submitqueue/entity/status.go b/submitqueue/entity/status.go deleted file mode 100644 index 3c6fe6de..00000000 --- a/submitqueue/entity/status.go +++ /dev/null @@ -1,21 +0,0 @@ -// 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 - -// StatusRequest identifies a request whose current status is being queried. -type StatusRequest struct { - // ID is the globally unique identifier of the request. Format: "/". - ID string -} diff --git a/submitqueue/gateway/controller/BUILD.bazel b/submitqueue/gateway/controller/BUILD.bazel index e07126c6..c71a3d4f 100644 --- a/submitqueue/gateway/controller/BUILD.bazel +++ b/submitqueue/gateway/controller/BUILD.bazel @@ -7,7 +7,7 @@ go_library( "land.go", "ping.go", "read_errors.go", - "status.go", + "request_summary.go", ], importpath = "github.com/uber/submitqueue/submitqueue/gateway/controller", visibility = ["//visibility:public"], @@ -34,7 +34,7 @@ go_test( "cancel_test.go", "land_test.go", "ping_test.go", - "status_test.go", + "request_summary_test.go", "storage_fixture_test.go", ], embed = [":go_default_library"], diff --git a/submitqueue/gateway/controller/read_errors.go b/submitqueue/gateway/controller/read_errors.go index feaa3e3a..861f3c92 100644 --- a/submitqueue/gateway/controller/read_errors.go +++ b/submitqueue/gateway/controller/read_errors.go @@ -14,11 +14,15 @@ package controller -import "fmt" +import ( + "errors" + "fmt" +) const ( maxQueueIdentifierBytes = 235 maxStorageIdentifierBytes = 255 + maxChangeRequestResults = 100 ) func validateQueueIdentifier(queue string) error { @@ -31,6 +35,59 @@ func validateQueueIdentifier(queue string) error { return nil } +// RequestNotFoundError indicates that no request exists for the selected sqid or change URI. +type RequestNotFoundError struct { + Sqid string + ChangeURI string +} + +// Error implements the error interface. +func (e *RequestNotFoundError) Error() string { + if e.Sqid != "" { + return fmt.Sprintf("request not found for sqid %q", e.Sqid) + } + return fmt.Sprintf("request not found for change URI %q", e.ChangeURI) +} + +// IsRequestNotFound returns true if any error in the chain is a *RequestNotFoundError. +func IsRequestNotFound(err error) bool { + var target *RequestNotFoundError + return errors.As(err, &target) +} + +// TooManyChangeRequestsError indicates that a change URI exceeded the API result limit. +type TooManyChangeRequestsError struct { + ChangeURI string + Limit int +} + +// Error implements the error interface. +func (e *TooManyChangeRequestsError) Error() string { + return fmt.Sprintf("change URI %q matched more than %d requests", e.ChangeURI, e.Limit) +} + +// IsTooManyChangeRequests returns true if any error in the chain is a *TooManyChangeRequestsError. +func IsTooManyChangeRequests(err error) bool { + var target *TooManyChangeRequestsError + return errors.As(err, &target) +} + +// InternalConsistencyError indicates that gateway-owned read models disagree. +type InternalConsistencyError struct { + Message string +} + +// Error implements the error interface. +func (e *InternalConsistencyError) Error() string { + return e.Message +} + +// IsInternalConsistency returns true if any error in the chain is an *InternalConsistencyError. +func IsInternalConsistency(err error) bool { + var target *InternalConsistencyError + return errors.As(err, &target) +} + func validateStoredIdentifier(name, value string) error { if value == "" { return fmt.Errorf("%s must be non-empty: %w", name, ErrInvalidRequest) diff --git a/submitqueue/gateway/controller/request_summary.go b/submitqueue/gateway/controller/request_summary.go new file mode 100644 index 00000000..ff604b19 --- /dev/null +++ b/submitqueue/gateway/controller/request_summary.go @@ -0,0 +1,111 @@ +// 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" + "fmt" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/errs" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" + "go.uber.org/zap" +) + +// RequestSummaryController handles materialized request-summary lookups for the gateway. +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{ + logger: logger, + metricsScope: scope.SubScope("request_summary_controller"), + requestSummaryStore: requestSummaryStore, + requestURIStore: requestURIStore, + } +} + +// GetRequestSummaryByID returns the current materialized view of one request. +func (c *RequestSummaryController) GetRequestSummaryByID(ctx context.Context, req entity.GetRequestSummaryByIDRequest) (summary entity.RequestSummary, retErr error) { + op := metrics.Begin(c.metricsScope, "get_by_id") + defer func() { op.Complete(retErr) }() + + if err := validateStoredIdentifier("sqid", req.ID); err != nil { + return entity.RequestSummary{}, fmt.Errorf("GetRequestSummaryByID invalid request: %w", err) + } + + summary, err := c.requestSummaryStore.Get(ctx, req.ID) + if err != nil { + if storage.IsNotFound(err) { + return entity.RequestSummary{}, errs.NewUserError(&RequestNotFoundError{Sqid: req.ID}) + } + return entity.RequestSummary{}, fmt.Errorf("GetRequestSummaryByID failed to get request summary sqid=%s: %w", req.ID, err) + } + if summary.Status == entity.RequestStatusAccepting { + return entity.RequestSummary{}, errs.NewUserError(&RequestNotFoundError{Sqid: req.ID}) + } + + c.logger.Debugw("request status retrieved", + "sqid", req.ID, + "status", string(summary.Status), + ) + return summary, nil +} + +// 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) { + op := metrics.Begin(c.metricsScope, "get_by_change_uri") + defer func() { op.Complete(retErr) }() + + if err := validateStoredIdentifier("change URI", req.ChangeURI); err != nil { + return nil, fmt.Errorf("GetRequestSummaryByChangeURI invalid request: %w", err) + } + + mappings, err := c.requestURIStore.ListByURI(ctx, req.ChangeURI, maxChangeRequestResults+1) + if err != nil { + return nil, fmt.Errorf("GetRequestSummaryByChangeURI failed to list request mappings change_uri=%s: %w", req.ChangeURI, err) + } + if len(mappings) == 0 { + return nil, errs.NewUserError(&RequestNotFoundError{ChangeURI: req.ChangeURI}) + } + if len(mappings) > maxChangeRequestResults { + return nil, errs.NewUserError(&TooManyChangeRequestsError{ChangeURI: req.ChangeURI, Limit: maxChangeRequestResults}) + } + + requests := make([]entity.RequestSummary, 0, len(mappings)) + for _, mapping := range mappings { + summary, err := c.requestSummaryStore.Get(ctx, mapping.RequestID) + if err != nil { + if storage.IsNotFound(err) { + return nil, &InternalConsistencyError{Message: fmt.Sprintf("request summary missing for mapped change URI %q and sqid %q", req.ChangeURI, mapping.RequestID)} + } + return nil, fmt.Errorf("GetRequestSummaryByChangeURI failed to get request summary change_uri=%s sqid=%s: %w", req.ChangeURI, mapping.RequestID, err) + } + requests = append(requests, summary) + } + + c.logger.Debugw("request statuses retrieved", + "change_uri", req.ChangeURI, + "request_count", len(requests), + ) + return requests, nil +} diff --git a/submitqueue/gateway/controller/request_summary_test.go b/submitqueue/gateway/controller/request_summary_test.go new file mode 100644 index 00000000..f54d78a8 --- /dev/null +++ b/submitqueue/gateway/controller/request_summary_test.go @@ -0,0 +1,206 @@ +// 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" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/errs" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" + storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +func TestGetRequestSummaryByID(t *testing.T) { + ctrl := gomock.NewController(t) + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + uriStore := storagemock.NewMockRequestURIStore(ctrl) + summaryStore.EXPECT().Get(gomock.Any(), "test-queue/1").Return(entity.RequestSummary{ + RequestID: "test-queue/1", + Queue: "test-queue", + ChangeURIs: []string{"github://uber/repo/pull/1/abc"}, + ReceivedAtMs: 100, + Status: entity.RequestStatusValidating, + LastError: "boom", + Metadata: map[string]string{"k": "v"}, + }, nil) + + controller := NewRequestSummaryController(zap.NewNop().Sugar(), tally.NoopScope, summaryStore, uriStore) + summary, err := controller.GetRequestSummaryByID(context.Background(), entity.GetRequestSummaryByIDRequest{ID: "test-queue/1"}) + + require.NoError(t, err) + assert.Equal(t, "test-queue/1", summary.RequestID) + assert.Equal(t, "test-queue", summary.Queue) + assert.Equal(t, []string{"github://uber/repo/pull/1/abc"}, summary.ChangeURIs) + assert.Equal(t, int64(100), summary.ReceivedAtMs) + assert.Equal(t, entity.RequestStatusValidating, summary.Status) + assert.Equal(t, "boom", summary.LastError) + assert.Equal(t, map[string]string{"k": "v"}, summary.Metadata) +} + +func TestGetRequestSummaryByChangeURI(t *testing.T) { + ctrl := gomock.NewController(t) + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + uriStore := storagemock.NewMockRequestURIStore(ctrl) + uriStore.EXPECT().ListByURI(gomock.Any(), "uri", 101).Return([]entity.RequestURI{ + {ChangeURI: "uri", ReceivedAtMs: 200, RequestID: "queue/2"}, + {ChangeURI: "uri", ReceivedAtMs: 100, RequestID: "queue/1"}, + }, nil) + summaryStore.EXPECT().Get(gomock.Any(), "queue/2").Return(entity.RequestSummary{RequestID: "queue/2", ReceivedAtMs: 200, Status: entity.RequestStatusLanded, ChangeURIs: []string{}}, nil) + summaryStore.EXPECT().Get(gomock.Any(), "queue/1").Return(entity.RequestSummary{RequestID: "queue/1", ReceivedAtMs: 100, Status: entity.RequestStatusError, ChangeURIs: []string{}}, nil) + + controller := NewRequestSummaryController(zap.NewNop().Sugar(), tally.NoopScope, summaryStore, uriStore) + summaries, err := controller.GetRequestSummaryByChangeURI(context.Background(), entity.GetRequestSummaryByChangeURIRequest{ChangeURI: "uri"}) + + require.NoError(t, err) + require.Len(t, summaries, 2) + assert.Equal(t, []string{"queue/2", "queue/1"}, []string{summaries[0].RequestID, summaries[1].RequestID}) +} + +func TestStatusErrors(t *testing.T) { + backendErr := fmt.Errorf("backend down") + tests := []struct { + name string + call func(*RequestSummaryController) error + setup func(*storagemock.MockRequestSummaryStore, *storagemock.MockRequestURIStore) + wantInvalid bool + wantNotFound bool + wantTooMany bool + wantInternal bool + wantUser bool + }{ + { + name: "empty sqid", + call: func(c *RequestSummaryController) error { + _, err := c.GetRequestSummaryByID(context.Background(), entity.GetRequestSummaryByIDRequest{}) + return err + }, + wantInvalid: true, + wantUser: true, + }, + { + name: "empty change URI", + call: func(c *RequestSummaryController) error { + _, err := c.GetRequestSummaryByChangeURI(context.Background(), entity.GetRequestSummaryByChangeURIRequest{}) + return err + }, + wantInvalid: true, + wantUser: true, + }, + { + name: "sqid not found", + setup: func(summaryStore *storagemock.MockRequestSummaryStore, _ *storagemock.MockRequestURIStore) { + summaryStore.EXPECT().Get(gomock.Any(), "missing/1").Return(entity.RequestSummary{}, storage.ErrNotFound) + }, + call: func(c *RequestSummaryController) error { + _, err := c.GetRequestSummaryByID(context.Background(), entity.GetRequestSummaryByIDRequest{ID: "missing/1"}) + return err + }, + wantNotFound: true, + wantUser: true, + }, + { + name: "sqid still accepting", + setup: func(summaryStore *storagemock.MockRequestSummaryStore, _ *storagemock.MockRequestURIStore) { + summaryStore.EXPECT().Get(gomock.Any(), "queue/1").Return(entity.RequestSummary{ + RequestID: "queue/1", + Status: entity.RequestStatusAccepting, + }, nil) + }, + call: func(c *RequestSummaryController) error { + _, err := c.GetRequestSummaryByID(context.Background(), entity.GetRequestSummaryByIDRequest{ID: "queue/1"}) + return err + }, + wantNotFound: true, + wantUser: true, + }, + { + name: "sqid storage failure", + setup: func(summaryStore *storagemock.MockRequestSummaryStore, _ *storagemock.MockRequestURIStore) { + summaryStore.EXPECT().Get(gomock.Any(), "queue/1").Return(entity.RequestSummary{}, backendErr) + }, + call: func(c *RequestSummaryController) error { + _, err := c.GetRequestSummaryByID(context.Background(), entity.GetRequestSummaryByIDRequest{ID: "queue/1"}) + return err + }, + }, + { + name: "change URI not found", + setup: func(_ *storagemock.MockRequestSummaryStore, uriStore *storagemock.MockRequestURIStore) { + uriStore.EXPECT().ListByURI(gomock.Any(), "uri", 101).Return([]entity.RequestURI{}, nil) + }, + call: func(c *RequestSummaryController) error { + _, err := c.GetRequestSummaryByChangeURI(context.Background(), entity.GetRequestSummaryByChangeURIRequest{ChangeURI: "uri"}) + return err + }, + wantNotFound: true, + wantUser: true, + }, + { + name: "too many change matches", + setup: func(_ *storagemock.MockRequestSummaryStore, uriStore *storagemock.MockRequestURIStore) { + uriStore.EXPECT().ListByURI(gomock.Any(), "uri", 101).Return(make([]entity.RequestURI, 101), nil) + }, + call: func(c *RequestSummaryController) error { + _, err := c.GetRequestSummaryByChangeURI(context.Background(), entity.GetRequestSummaryByChangeURIRequest{ChangeURI: "uri"}) + return err + }, + wantTooMany: true, + wantUser: true, + }, + { + name: "mapped summary missing", + setup: func(summaryStore *storagemock.MockRequestSummaryStore, uriStore *storagemock.MockRequestURIStore) { + uriStore.EXPECT().ListByURI(gomock.Any(), "uri", 101).Return([]entity.RequestURI{{RequestID: "missing/1"}}, nil) + summaryStore.EXPECT().Get(gomock.Any(), "missing/1").Return(entity.RequestSummary{}, storage.ErrNotFound) + }, + call: func(c *RequestSummaryController) error { + _, err := c.GetRequestSummaryByChangeURI(context.Background(), entity.GetRequestSummaryByChangeURIRequest{ChangeURI: "uri"}) + return err + }, + wantInternal: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + uriStore := storagemock.NewMockRequestURIStore(ctrl) + if tt.setup != nil { + tt.setup(summaryStore, uriStore) + } + controller := NewRequestSummaryController(zap.NewNop().Sugar(), tally.NoopScope, summaryStore, uriStore) + + err := tt.call(controller) + + require.Error(t, err) + if tt.wantInvalid { + assert.True(t, IsInvalidRequest(err)) + } + assert.Equal(t, tt.wantNotFound, IsRequestNotFound(err)) + assert.Equal(t, tt.wantTooMany, IsTooManyChangeRequests(err)) + assert.Equal(t, tt.wantInternal, IsInternalConsistency(err)) + assert.Equal(t, tt.wantUser, errs.IsUserError(err)) + }) + } +} diff --git a/submitqueue/gateway/controller/status.go b/submitqueue/gateway/controller/status.go deleted file mode 100644 index ea6c04d4..00000000 --- a/submitqueue/gateway/controller/status.go +++ /dev/null @@ -1,98 +0,0 @@ -// 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" - "time" - - "github.com/uber-go/tally" - "github.com/uber/submitqueue/platform/errs" - "github.com/uber/submitqueue/submitqueue/core/request" - "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" - "go.uber.org/zap" -) - -// RequestNotFoundError indicates that no request log records exist for the -// requested sqid. Either the sqid is wrong or the request has not been -// accepted yet. -type RequestNotFoundError struct { - Sqid string -} - -// Error implements the error interface. -func (e *RequestNotFoundError) Error() string { - return fmt.Sprintf("request not found for sqid %q", e.Sqid) -} - -// IsRequestNotFound returns true if any error in the chain is a -// *RequestNotFoundError. -func IsRequestNotFound(err error) bool { - var target *RequestNotFoundError - return errors.As(err, &target) -} - -// StatusController handles request status business logic for the gateway. -type StatusController struct { - logger *zap.SugaredLogger - metricsScope tally.Scope - requestLogStore storage.RequestLogStore -} - -// NewStatusController creates a new instance of the gateway status controller. -func NewStatusController(logger *zap.SugaredLogger, scope tally.Scope, requestLogStore storage.RequestLogStore) *StatusController { - return &StatusController{ - logger: logger, - metricsScope: scope, - requestLogStore: requestLogStore, - } -} - -// StatusResult is the outcome of a status query. It is a type alias for -// request.CurrentState so that the controller's return type follows the -// same Result naming convention as LandResult and other controller outputs. -type StatusResult = request.CurrentState - -// Status returns the current reconciled status of a request identified by its sqid. -func (c *StatusController) Status(ctx context.Context, req entity.StatusRequest) (StatusResult, error) { - start := time.Now() - defer func() { - c.metricsScope.Timer("status_latency").Record(time.Since(start)) - }() - - c.metricsScope.Counter("status_count").Inc(1) - - if req.ID == "" { - return StatusResult{}, fmt.Errorf("StatusController requires the request to have a sqid specified: %w", ErrInvalidRequest) - } - - state, err := request.GetCurrentStateFromRequestLog(ctx, c.requestLogStore, req.ID) - if err != nil { - if storage.IsNotFound(err) { - return StatusResult{}, errs.NewUserError(&RequestNotFoundError{Sqid: req.ID}) - } - return StatusResult{}, fmt.Errorf("StatusController failed to get current state for sqid=%s: %w", req.ID, err) - } - - c.logger.Debugw("request status retrieved", - "sqid", req.ID, - "status", string(state.Status), - ) - - return state, nil -} diff --git a/submitqueue/gateway/controller/status_test.go b/submitqueue/gateway/controller/status_test.go deleted file mode 100644 index 3791f059..00000000 --- a/submitqueue/gateway/controller/status_test.go +++ /dev/null @@ -1,112 +0,0 @@ -// 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" - "fmt" - "testing" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - "github.com/uber-go/tally" - "github.com/uber/submitqueue/platform/errs" - "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" - storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" - "go.uber.org/mock/gomock" - "go.uber.org/zap" -) - -func TestStatus_ReturnsCurrentState(t *testing.T) { - ctrl := gomock.NewController(t) - - store := storagemock.NewMockRequestLogStore(ctrl) - store.EXPECT().List(gomock.Any(), "test-queue/1").Return([]entity.RequestLog{ - {RequestID: "test-queue/1", TimestampMs: 100, Status: entity.RequestStatusAccepted}, - {RequestID: "test-queue/1", TimestampMs: 200, Status: entity.RequestStatusValidating, LastError: "boom", Metadata: map[string]string{"k": "v"}}, - }, nil) - - controller := NewStatusController(zap.NewNop().Sugar(), tally.NoopScope, store) - - state, err := controller.Status(context.Background(), entity.StatusRequest{ID: "test-queue/1"}) - - require.NoError(t, err) - assert.Equal(t, entity.RequestStatusValidating, state.Status) - assert.Equal(t, "boom", state.LastError) - assert.Equal(t, map[string]string{"k": "v"}, state.Metadata) -} - -func TestStatus_Errors(t *testing.T) { - tests := []struct { - name string - sqid string - setupStore func(*storagemock.MockRequestLogStore) - wantInvalid bool - wantNotFound bool - wantUserError bool - }{ - { - name: "empty sqid is an invalid request", - sqid: "", - wantInvalid: true, - wantUserError: true, - }, - { - name: "unknown sqid maps to not found", - sqid: "missing/1", - setupStore: func(s *storagemock.MockRequestLogStore) { - s.EXPECT().List(gomock.Any(), "missing/1").Return(nil, storage.ErrNotFound) - }, - wantNotFound: true, - wantUserError: true, - }, - { - name: "store failure propagates as infra error", - sqid: "test-queue/1", - setupStore: func(s *storagemock.MockRequestLogStore) { - s.EXPECT().List(gomock.Any(), "test-queue/1").Return(nil, fmt.Errorf("log backend down")) - }, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - ctrl := gomock.NewController(t) - store := storagemock.NewMockRequestLogStore(ctrl) - if tt.setupStore != nil { - tt.setupStore(store) - } - - controller := NewStatusController(zap.NewNop().Sugar(), tally.NoopScope, store) - - _, err := controller.Status(context.Background(), entity.StatusRequest{ID: tt.sqid}) - - require.Error(t, err) - assert.Equal(t, tt.wantNotFound, IsRequestNotFound(err)) - assert.Equal(t, tt.wantUserError, errs.IsUserError(err)) - assert.False(t, errs.IsRetryable(err)) - if tt.wantInvalid { - assert.True(t, IsInvalidRequest(err)) - } - - if tt.wantNotFound { - var typed *RequestNotFoundError - require.ErrorAs(t, err, &typed) - assert.Equal(t, tt.sqid, typed.Sqid) - } - }) - } -} diff --git a/test/e2e/submitqueue/harness_test.go b/test/e2e/submitqueue/harness_test.go index b8d18eef..81094004 100644 --- a/test/e2e/submitqueue/harness_test.go +++ b/test/e2e/submitqueue/harness_test.go @@ -15,10 +15,10 @@ package e2e_test // Reusable e2e helpers so tests read as intent, not plumbing. They drive the -// stack through the real gateway gRPC surface (Land / Cancel / Status) and +// stack through the real gateway gRPC surface (Land / Cancel / GetRequestSummaryByID) and // observe outcomes two ways: // -// - black-box, by polling the Status RPC to a target/terminal status; and +// - black-box, by polling the GetRequestSummaryByID RPC to a target/terminal status; and // - white-box, by reading the request_log timeline (RequestLogStore.List on // mysql-app) to assert the ordered stage progression. // @@ -28,6 +28,8 @@ package e2e_test // stage is genuinely stuck, not a timing race. import ( + "fmt" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" changepb "github.com/uber/submitqueue/api/base/change/protopb" @@ -51,32 +53,35 @@ func (s *E2EIntegrationSuite) land(queue string, uris ...string) string { return resp.Sqid } -// currentStatus reads the request's current customer-facing status via the -// Status RPC. A transport error is returned so callers can keep polling. +// currentStatus reads the request's current customer-facing status via +// GetRequestSummaryByID. A transport error is returned so callers can keep polling. func (s *E2EIntegrationSuite) currentStatus(sqid string) (entity.RequestStatus, error) { - resp, err := s.gatewayClient.Status(s.ctx, &gatewaypb.StatusRequest{Sqid: sqid}) + resp, err := s.gatewayClient.GetRequestSummaryByID(s.ctx, &gatewaypb.GetRequestSummaryByIDRequest{Sqid: sqid}) if err != nil { return entity.RequestStatusUnknown, err } - return entity.RequestStatus(resp.Status), nil + if resp.Request == nil { + return entity.RequestStatusUnknown, fmt.Errorf("GetRequestSummaryByID(%s) returned no request", sqid) + } + return entity.RequestStatus(resp.Request.Status), nil } -// awaitStatus polls Status until the request reaches exactly want. +// awaitStatus polls GetRequestSummaryByID until the request reaches exactly want. func (s *E2EIntegrationSuite) awaitStatus(sqid string, want entity.RequestStatus) { t := s.T() require.Eventually(t, func() bool { got, err := s.currentStatus(sqid) if err != nil { - s.log.Logf("Status(%s) not ready yet: %v", sqid, err) + s.log.Logf("GetRequestSummaryByID(%s) not ready yet: %v", sqid, err) return false } - s.log.Logf("Status(%s) = %q (want %q)", sqid, got, want) + s.log.Logf("GetRequestSummaryByID(%s) = %q (want %q)", sqid, got, want) return got == want }, persistTimeout, persistPollInterval, "request %s should reach status %q", sqid, want) } -// awaitTerminal polls Status until the request reaches a terminal status +// awaitTerminal polls GetRequestSummaryByID until the request reaches a terminal status // (landed, error, or cancelled) and returns it. func (s *E2EIntegrationSuite) awaitTerminal(sqid string) entity.RequestStatus { t := s.T() @@ -84,19 +89,17 @@ func (s *E2EIntegrationSuite) awaitTerminal(sqid string) entity.RequestStatus { require.Eventually(t, func() bool { got, err := s.currentStatus(sqid) if err != nil { - s.log.Logf("Status(%s) not ready yet: %v", sqid, err) + s.log.Logf("GetRequestSummaryByID(%s) not ready yet: %v", sqid, err) return false } last = got - s.log.Logf("Status(%s) = %q (awaiting terminal)", sqid, got) + s.log.Logf("GetRequestSummaryByID(%s) = %q (awaiting terminal)", sqid, got) return isTerminalStatus(got) }, persistTimeout, persistPollInterval, "request %s should reach a terminal status", sqid) return last } -// timeline returns the ordered status history from the request_log (the audit -// trail persisted by the gateway log consumer on mysql-app). // timeline returns the ordered status history from the request_log (the audit // trail persisted by the gateway log consumer on mysql-app). These are the // customer-facing RequestStatus values — the only ordered history in the system @@ -141,13 +144,13 @@ func (s *E2EIntegrationSuite) terminalState(sqid string) entity.RequestState { return req.State } -// lastError returns the LastError reported by the Status RPC (populated on the -// error path). +// lastError returns the LastError reported by GetRequestSummaryByID. func (s *E2EIntegrationSuite) lastError(sqid string) string { t := s.T() - resp, err := s.gatewayClient.Status(s.ctx, &gatewaypb.StatusRequest{Sqid: sqid}) - require.NoError(t, err, "Status failed for %s", sqid) - return resp.LastError + resp, err := s.gatewayClient.GetRequestSummaryByID(s.ctx, &gatewaypb.GetRequestSummaryByIDRequest{Sqid: sqid}) + require.NoError(t, err, "GetRequestSummaryByID failed for %s", sqid) + require.NotNil(t, resp.Request) + return resp.Request.LastError } // isTerminalStatus reports whether a customer-facing status is terminal. diff --git a/test/integration/submitqueue/gateway/BUILD.bazel b/test/integration/submitqueue/gateway/BUILD.bazel index 36fda2d6..34fa0ad0 100644 --- a/test/integration/submitqueue/gateway/BUILD.bazel +++ b/test/integration/submitqueue/gateway/BUILD.bazel @@ -31,6 +31,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_uber_go_zap//:go_default_library", ], ) diff --git a/test/integration/submitqueue/gateway/suite_test.go b/test/integration/submitqueue/gateway/suite_test.go index 1c0c3899..cbf087b5 100644 --- a/test/integration/submitqueue/gateway/suite_test.go +++ b/test/integration/submitqueue/gateway/suite_test.go @@ -28,6 +28,7 @@ package gateway import ( "context" "database/sql" + "fmt" "path/filepath" "testing" "time" @@ -48,6 +49,8 @@ import ( "github.com/uber/submitqueue/test/testutil" "go.uber.org/zap" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) type GatewayIntegrationSuite struct { @@ -65,7 +68,7 @@ func TestGatewayIntegration(t *testing.T) { } // The log consumer runs inside the gateway-service container, so this suite can -// only observe persistence black-box through the Status RPC — there is no +// only observe persistence black-box through the request-summary RPC — there is no // in-process channel/HookSignal to wait on across the container boundary. A // bounded poll is therefore the deterministic-enough analog: persistTimeout is a // safety net (a failure here means something is genuinely stuck, not a timing @@ -167,12 +170,46 @@ func (s *GatewayIntegrationSuite) TestLandAPI() { s.log.Logf("Land API test passed: request stored and message published") } +// TestRequestSummaryAPIErrorCodes verifies request-summary controller errors reach stable gRPC codes. +func (s *GatewayIntegrationSuite) TestRequestSummaryAPIErrorCodes() { + t := s.T() + + _, err := s.client.GetRequestSummaryByID(s.ctx, &pb.GetRequestSummaryByIDRequest{Sqid: "missing/1"}) + require.Error(t, err) + assert.Equal(t, codes.NotFound, status.Code(err)) + + store, err := mysqlstorage.NewStorage(s.db, tally.NoopScope) + require.NoError(t, err) + const overflowChangeURI = "uri/read-api-overflow" + for i := 1; i <= 101; i++ { + require.NoError(t, store.GetRequestURIStore().Create(s.ctx, entity.RequestURI{ + ChangeURI: overflowChangeURI, + ReceivedAtMs: int64(i), + RequestID: fmt.Sprintf("overflow/%d", i), + })) + } + + _, err = s.client.GetRequestSummaryByChangeURI(s.ctx, &pb.GetRequestSummaryByChangeURIRequest{ChangeUri: overflowChangeURI}) + require.Error(t, err) + assert.Equal(t, codes.ResourceExhausted, status.Code(err)) + + const inconsistentChangeURI = "uri/read-api-inconsistent" + require.NoError(t, store.GetRequestURIStore().Create(s.ctx, entity.RequestURI{ + ChangeURI: inconsistentChangeURI, + ReceivedAtMs: 1, + RequestID: "missing-summary/1", + })) + _, err = s.client.GetRequestSummaryByChangeURI(s.ctx, &pb.GetRequestSummaryByChangeURIRequest{ChangeUri: inconsistentChangeURI}) + require.Error(t, err) + assert.Equal(t, codes.Internal, status.Code(err)) +} + // TestRequestLogConsumer verifies the gateway's log-topic consumer in isolation: // no orchestrator runs in this stack, so the test itself publishes a request log // entry to the log topic exactly as the orchestrator does in production (via // submitqueue/core/request.PublishLog). The gateway is the sole writer of the // request log; this asserts its consumer drains the log topic and persists the -// entry to storage, observable through the Status RPC. +// entry to storage, observable through the request-summary RPC. func (s *GatewayIntegrationSuite) TestRequestLogConsumer() { t := s.T() @@ -207,13 +244,13 @@ func (s *GatewayIntegrationSuite) TestRequestLogConsumer() { s.log.Logf("Published 'started' log for sqid=%s; waiting for gateway consumer to persist it", sqid) require.Eventually(t, func() bool { - resp, statusErr := s.client.Status(s.ctx, &pb.StatusRequest{Sqid: sqid}) + resp, statusErr := s.client.GetRequestSummaryByID(s.ctx, &pb.GetRequestSummaryByIDRequest{Sqid: sqid}) if statusErr != nil { return false } - return resp.Status == string(entity.RequestStatusStarted) + return resp.Request != nil && resp.Request.Status == string(entity.RequestStatusStarted) }, persistTimeout, persistPollInterval, "gateway log consumer should persist the published request log for sqid=%s", sqid) - s.log.Logf("Request log consumer test passed: entry persisted and readable via Status") + s.log.Logf("Request log consumer test passed: entry persisted and readable via GetRequestSummaryByID") }