mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
## What changed? Add batch workflow refresh tasks Admin API. Add tdbg commands to invoke `StartAdminBatchOperation`. ## Details AdminBatchOperations are executed in the Batched System Worker, similar to existing Frontend BatchOperations. The worker activity will fetch pages of workflow executions to perform operations on, and periodically heartbeat its results. For RefreshWorkflowTasks, each individual Refresh call will go through the admin handler. Refreshing Workflow Tasks regenerates all pending tasks of an execution given its mutable state. ## CLI usage ``` tdbg workflow batch-refresh-tasks \ --namespace <ns> \ --query "WorkflowType='MyType'" \ --reason "fixing stuck workflows" \ [--job-id <optional-job-id>] \ [--archetype <optional-archetype>] ``` ## Why? Unblock new Matcher migration and allow for general use case batch Admin calls. Currently, only supports `BatchOperationRefreshWorkflowTasks`. ## How did you test it? - [X] built - [X] run locally and tested manually - [X] covered by existing tests - [X] added new unit test(s) - [X] added new functional test(s)
205 lines
9.0 KiB
Go
205 lines
9.0 KiB
Go
// Code generated by protoc-gen-go. DO NOT EDIT.
|
|
// plugins:
|
|
// protoc-gen-go
|
|
// protoc
|
|
// source: temporal/server/api/batch/v1/request_response.proto
|
|
|
|
package batch
|
|
|
|
import (
|
|
reflect "reflect"
|
|
sync "sync"
|
|
unsafe "unsafe"
|
|
|
|
v1 "go.temporal.io/api/enums/v1"
|
|
v11 "go.temporal.io/api/workflowservice/v1"
|
|
v12 "go.temporal.io/server/api/adminservice/v1"
|
|
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
|
|
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
|
|
durationpb "google.golang.org/protobuf/types/known/durationpb"
|
|
)
|
|
|
|
const (
|
|
// Verify that this generated code is sufficiently up-to-date.
|
|
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
|
|
// Verify that runtime/protoimpl is sufficiently up-to-date.
|
|
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
|
|
)
|
|
|
|
type BatchOperationInput struct {
|
|
state protoimpl.MessageState `protogen:"open.v1"`
|
|
NamespaceId string `protobuf:"bytes,1,opt,name=namespace_id,json=namespaceId,proto3" json:"namespace_id,omitempty"`
|
|
Concurrency int64 `protobuf:"varint,2,opt,name=concurrency,proto3" json:"concurrency,omitempty"`
|
|
AttemptsOnRetryableError int64 `protobuf:"varint,3,opt,name=attempts_on_retryable_error,json=attemptsOnRetryableError,proto3" json:"attempts_on_retryable_error,omitempty"`
|
|
ActivityHeartbeatTimeout *durationpb.Duration `protobuf:"bytes,4,opt,name=activity_heartbeat_timeout,json=activityHeartbeatTimeout,proto3" json:"activity_heartbeat_timeout,omitempty"`
|
|
NonRetryableErrors []string `protobuf:"bytes,5,rep,name=non_retryable_errors,json=nonRetryableErrors,proto3" json:"non_retryable_errors,omitempty"`
|
|
// Only needed if StartBatchOperationRequest request is set.
|
|
BatchType v1.BatchOperationType `protobuf:"varint,6,opt,name=batch_type,json=batchType,proto3,enum=temporal.api.enums.v1.BatchOperationType" json:"batch_type,omitempty"`
|
|
// The request to start the batch operation.
|
|
// Mutually exclusive with StartAdminBatchOperationRequest admin_request.
|
|
Request *v11.StartBatchOperationRequest `protobuf:"bytes,7,opt,name=request,proto3" json:"request,omitempty"`
|
|
// The request to start an admin batch operation.
|
|
// Mutually exclusive with StartBatchOperationRequest request.
|
|
AdminRequest *v12.StartAdminBatchOperationRequest `protobuf:"bytes,8,opt,name=admin_request,json=adminRequest,proto3" json:"admin_request,omitempty"`
|
|
unknownFields protoimpl.UnknownFields
|
|
sizeCache protoimpl.SizeCache
|
|
}
|
|
|
|
func (x *BatchOperationInput) Reset() {
|
|
*x = BatchOperationInput{}
|
|
mi := &file_temporal_server_api_batch_v1_request_response_proto_msgTypes[0]
|
|
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
|
ms.StoreMessageInfo(mi)
|
|
}
|
|
|
|
func (x *BatchOperationInput) String() string {
|
|
return protoimpl.X.MessageStringOf(x)
|
|
}
|
|
|
|
func (*BatchOperationInput) ProtoMessage() {}
|
|
|
|
func (x *BatchOperationInput) ProtoReflect() protoreflect.Message {
|
|
mi := &file_temporal_server_api_batch_v1_request_response_proto_msgTypes[0]
|
|
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 BatchOperationInput.ProtoReflect.Descriptor instead.
|
|
func (*BatchOperationInput) Descriptor() ([]byte, []int) {
|
|
return file_temporal_server_api_batch_v1_request_response_proto_rawDescGZIP(), []int{0}
|
|
}
|
|
|
|
func (x *BatchOperationInput) GetNamespaceId() string {
|
|
if x != nil {
|
|
return x.NamespaceId
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func (x *BatchOperationInput) GetConcurrency() int64 {
|
|
if x != nil {
|
|
return x.Concurrency
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func (x *BatchOperationInput) GetAttemptsOnRetryableError() int64 {
|
|
if x != nil {
|
|
return x.AttemptsOnRetryableError
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func (x *BatchOperationInput) GetActivityHeartbeatTimeout() *durationpb.Duration {
|
|
if x != nil {
|
|
return x.ActivityHeartbeatTimeout
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (x *BatchOperationInput) GetNonRetryableErrors() []string {
|
|
if x != nil {
|
|
return x.NonRetryableErrors
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (x *BatchOperationInput) GetBatchType() v1.BatchOperationType {
|
|
if x != nil {
|
|
return x.BatchType
|
|
}
|
|
return v1.BatchOperationType(0)
|
|
}
|
|
|
|
func (x *BatchOperationInput) GetRequest() *v11.StartBatchOperationRequest {
|
|
if x != nil {
|
|
return x.Request
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (x *BatchOperationInput) GetAdminRequest() *v12.StartAdminBatchOperationRequest {
|
|
if x != nil {
|
|
return x.AdminRequest
|
|
}
|
|
return nil
|
|
}
|
|
|
|
var File_temporal_server_api_batch_v1_request_response_proto protoreflect.FileDescriptor
|
|
|
|
const file_temporal_server_api_batch_v1_request_response_proto_rawDesc = "" +
|
|
"\n" +
|
|
"3temporal/server/api/batch/v1/request_response.proto\x12\x1ctemporal.server.api.batch.v1\x1a6temporal/api/workflowservice/v1/request_response.proto\x1a+temporal/api/enums/v1/batch_operation.proto\x1a:temporal/server/api/adminservice/v1/request_response.proto\x1a\x1egoogle/protobuf/duration.proto\"\xb0\x04\n" +
|
|
"\x13BatchOperationInput\x12!\n" +
|
|
"\fnamespace_id\x18\x01 \x01(\tR\vnamespaceId\x12 \n" +
|
|
"\vconcurrency\x18\x02 \x01(\x03R\vconcurrency\x12=\n" +
|
|
"\x1battempts_on_retryable_error\x18\x03 \x01(\x03R\x18attemptsOnRetryableError\x12W\n" +
|
|
"\x1aactivity_heartbeat_timeout\x18\x04 \x01(\v2\x19.google.protobuf.DurationR\x18activityHeartbeatTimeout\x120\n" +
|
|
"\x14non_retryable_errors\x18\x05 \x03(\tR\x12nonRetryableErrors\x12H\n" +
|
|
"\n" +
|
|
"batch_type\x18\x06 \x01(\x0e2).temporal.api.enums.v1.BatchOperationTypeR\tbatchType\x12U\n" +
|
|
"\arequest\x18\a \x01(\v2;.temporal.api.workflowservice.v1.StartBatchOperationRequestR\arequest\x12i\n" +
|
|
"\radmin_request\x18\b \x01(\v2D.temporal.server.api.adminservice.v1.StartAdminBatchOperationRequestR\fadminRequestB*Z(go.temporal.io/server/api/batch/v1;batchb\x06proto3"
|
|
|
|
var (
|
|
file_temporal_server_api_batch_v1_request_response_proto_rawDescOnce sync.Once
|
|
file_temporal_server_api_batch_v1_request_response_proto_rawDescData []byte
|
|
)
|
|
|
|
func file_temporal_server_api_batch_v1_request_response_proto_rawDescGZIP() []byte {
|
|
file_temporal_server_api_batch_v1_request_response_proto_rawDescOnce.Do(func() {
|
|
file_temporal_server_api_batch_v1_request_response_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_temporal_server_api_batch_v1_request_response_proto_rawDesc), len(file_temporal_server_api_batch_v1_request_response_proto_rawDesc)))
|
|
})
|
|
return file_temporal_server_api_batch_v1_request_response_proto_rawDescData
|
|
}
|
|
|
|
var file_temporal_server_api_batch_v1_request_response_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
|
|
var file_temporal_server_api_batch_v1_request_response_proto_goTypes = []any{
|
|
(*BatchOperationInput)(nil), // 0: temporal.server.api.batch.v1.BatchOperationInput
|
|
(*durationpb.Duration)(nil), // 1: google.protobuf.Duration
|
|
(v1.BatchOperationType)(0), // 2: temporal.api.enums.v1.BatchOperationType
|
|
(*v11.StartBatchOperationRequest)(nil), // 3: temporal.api.workflowservice.v1.StartBatchOperationRequest
|
|
(*v12.StartAdminBatchOperationRequest)(nil), // 4: temporal.server.api.adminservice.v1.StartAdminBatchOperationRequest
|
|
}
|
|
var file_temporal_server_api_batch_v1_request_response_proto_depIdxs = []int32{
|
|
1, // 0: temporal.server.api.batch.v1.BatchOperationInput.activity_heartbeat_timeout:type_name -> google.protobuf.Duration
|
|
2, // 1: temporal.server.api.batch.v1.BatchOperationInput.batch_type:type_name -> temporal.api.enums.v1.BatchOperationType
|
|
3, // 2: temporal.server.api.batch.v1.BatchOperationInput.request:type_name -> temporal.api.workflowservice.v1.StartBatchOperationRequest
|
|
4, // 3: temporal.server.api.batch.v1.BatchOperationInput.admin_request:type_name -> temporal.server.api.adminservice.v1.StartAdminBatchOperationRequest
|
|
4, // [4:4] is the sub-list for method output_type
|
|
4, // [4:4] is the sub-list for method input_type
|
|
4, // [4:4] is the sub-list for extension type_name
|
|
4, // [4:4] is the sub-list for extension extendee
|
|
0, // [0:4] is the sub-list for field type_name
|
|
}
|
|
|
|
func init() { file_temporal_server_api_batch_v1_request_response_proto_init() }
|
|
func file_temporal_server_api_batch_v1_request_response_proto_init() {
|
|
if File_temporal_server_api_batch_v1_request_response_proto != nil {
|
|
return
|
|
}
|
|
type x struct{}
|
|
out := protoimpl.TypeBuilder{
|
|
File: protoimpl.DescBuilder{
|
|
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
|
|
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_batch_v1_request_response_proto_rawDesc), len(file_temporal_server_api_batch_v1_request_response_proto_rawDesc)),
|
|
NumEnums: 0,
|
|
NumMessages: 1,
|
|
NumExtensions: 0,
|
|
NumServices: 0,
|
|
},
|
|
GoTypes: file_temporal_server_api_batch_v1_request_response_proto_goTypes,
|
|
DependencyIndexes: file_temporal_server_api_batch_v1_request_response_proto_depIdxs,
|
|
MessageInfos: file_temporal_server_api_batch_v1_request_response_proto_msgTypes,
|
|
}.Build()
|
|
File_temporal_server_api_batch_v1_request_response_proto = out.File
|
|
file_temporal_server_api_batch_v1_request_response_proto_goTypes = nil
|
|
file_temporal_server_api_batch_v1_request_response_proto_depIdxs = nil
|
|
}
|