Files
tasks/api/batch/v1/request_response.pb.go
Alan WuandGitHub cf7c62c8cc Add batch workflow refresh tasks (#8793)
## 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)
2026-01-06 19:21:57 -05:00

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
}