Github/temporal
1
0
Fork 0
mirror of https://github.com/temporalio/temporal.git synced 2026-08-30 18:41:49 -07:00
Code Issues Packages Projects Releases Wiki Activity

Revert raw history changes (#7314)

Browse Source 
- **Revert "Fix history batch validations (#7281)"**
- **Revert "Send raw history blobs from history service to frontend
(#7179)"**

## What changed?
Revert the changes for returning raw history event from history to
frontend.

## Why?
Some logic is moved from history to frontend handler. This might cause
issues during upgrade.
I will push a change to first copy these operations to frontend. Once
that change is released, I will reapply this change.

## How did you test it?

## Potential risks

## Documentation

## Is hotfix candidate?
This commit is contained in:
Prathyush PV
2025-02-11 10:13:13 -08:00
committed by GitHub
parent 07196d2971
commit 4bd66830e2
25 changed files with 2835 additions and 3673 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
api/history/v1/message.go-helpers.pb.go
View File

@@ -250,114 +250,3 @@ func (this *TaskRange) Equal(that interface{}) bool {
return proto.Equal(this, that1)
}
// Marshal an object of type StrippedHistoryEvent to the protobuf v3 wire format
func (val *StrippedHistoryEvent) Marshal() ([]byte, error) {
return proto.Marshal(val)
}
// Unmarshal an object of type StrippedHistoryEvent from the protobuf v3 wire format
func (val *StrippedHistoryEvent) Unmarshal(buf []byte) error {
return proto.Unmarshal(buf, val)
}
// Size returns the size of the object, in bytes, once serialized
func (val *StrippedHistoryEvent) Size() int {
return proto.Size(val)
}
// Equal returns whether two StrippedHistoryEvent values are equivalent by recursively
// comparing the message's fields.
// For more information see the documentation for
// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
func (this *StrippedHistoryEvent) Equal(that interface{}) bool {
if that == nil {
return this == nil
}
var that1 *StrippedHistoryEvent
switch t := that.(type) {
case *StrippedHistoryEvent:
that1 = t
case StrippedHistoryEvent:
that1 = &t
default:
return false
}
return proto.Equal(this, that1)
}
// Marshal an object of type StrippedHistoryEvents to the protobuf v3 wire format
func (val *StrippedHistoryEvents) Marshal() ([]byte, error) {
return proto.Marshal(val)
}
// Unmarshal an object of type StrippedHistoryEvents from the protobuf v3 wire format
func (val *StrippedHistoryEvents) Unmarshal(buf []byte) error {
return proto.Unmarshal(buf, val)
}
// Size returns the size of the object, in bytes, once serialized
func (val *StrippedHistoryEvents) Size() int {
return proto.Size(val)
}
// Equal returns whether two StrippedHistoryEvents values are equivalent by recursively
// comparing the message's fields.
// For more information see the documentation for
// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
func (this *StrippedHistoryEvents) Equal(that interface{}) bool {
if that == nil {
return this == nil
}
var that1 *StrippedHistoryEvents
switch t := that.(type) {
case *StrippedHistoryEvents:
that1 = t
case StrippedHistoryEvents:
that1 = &t
default:
return false
}
return proto.Equal(this, that1)
}
// Marshal an object of type GetWorkflowExecutionHistoryResponse to the protobuf v3 wire format
func (val *GetWorkflowExecutionHistoryResponse) Marshal() ([]byte, error) {
return proto.Marshal(val)
}
// Unmarshal an object of type GetWorkflowExecutionHistoryResponse from the protobuf v3 wire format
func (val *GetWorkflowExecutionHistoryResponse) Unmarshal(buf []byte) error {
return proto.Unmarshal(buf, val)
}
// Size returns the size of the object, in bytes, once serialized
func (val *GetWorkflowExecutionHistoryResponse) Size() int {
return proto.Size(val)
}
// Equal returns whether two GetWorkflowExecutionHistoryResponse values are equivalent by recursively
// comparing the message's fields.
// For more information see the documentation for
// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
func (this *GetWorkflowExecutionHistoryResponse) Equal(that interface{}) bool {
if that == nil {
return this == nil
}
var that1 *GetWorkflowExecutionHistoryResponse
switch t := that.(type) {
case *GetWorkflowExecutionHistoryResponse:
that1 = t
case GetWorkflowExecutionHistoryResponse:
that1 = &t
default:
return false
}
return proto.Equal(this, that1)
}

393
api/history/v1/message.pb.go
View File

@@ -32,7 +32,6 @@ import (
reflect "reflect"
sync "sync"
v11 "go.temporal.io/api/common/v1"
v1 "go.temporal.io/api/history/v1"
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
@@ -372,277 +371,71 @@ func (x *TaskRange) GetExclusiveMaxTaskKey() *TaskKey {
return nil
}
// StrippedHistoryEvent is a stripped down version of HistoryEvent that only contains the event_id and version.
type StrippedHistoryEvent struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
EventId int64 `protobuf:"varint,1,opt,name=event_id,json=eventId,proto3" json:"event_id,omitempty"`
Version int64 `protobuf:"varint,4,opt,name=version,proto3" json:"version,omitempty"`
}
func (x *StrippedHistoryEvent) Reset() {
*x = StrippedHistoryEvent{}
if protoimpl.UnsafeEnabled {
mi := &file_temporal_server_api_history_v1_message_proto_msgTypes[6]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *StrippedHistoryEvent) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*StrippedHistoryEvent) ProtoMessage() {}
func (x *StrippedHistoryEvent) ProtoReflect() protoreflect.Message {
mi := &file_temporal_server_api_history_v1_message_proto_msgTypes[6]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use StrippedHistoryEvent.ProtoReflect.Descriptor instead.
func (*StrippedHistoryEvent) Descriptor() ([]byte, []int) {
return file_temporal_server_api_history_v1_message_proto_rawDescGZIP(), []int{6}
}
func (x *StrippedHistoryEvent) GetEventId() int64 {
if x != nil {
return x.EventId
}
return 0
}
func (x *StrippedHistoryEvent) GetVersion() int64 {
if x != nil {
return x.Version
}
return 0
}
type StrippedHistoryEvents struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Events []*StrippedHistoryEvent `protobuf:"bytes,1,rep,name=events,proto3" json:"events,omitempty"`
}
func (x *StrippedHistoryEvents) Reset() {
*x = StrippedHistoryEvents{}
if protoimpl.UnsafeEnabled {
mi := &file_temporal_server_api_history_v1_message_proto_msgTypes[7]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *StrippedHistoryEvents) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*StrippedHistoryEvents) ProtoMessage() {}
func (x *StrippedHistoryEvents) ProtoReflect() protoreflect.Message {
mi := &file_temporal_server_api_history_v1_message_proto_msgTypes[7]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use StrippedHistoryEvents.ProtoReflect.Descriptor instead.
func (*StrippedHistoryEvents) Descriptor() ([]byte, []int) {
return file_temporal_server_api_history_v1_message_proto_rawDescGZIP(), []int{7}
}
func (x *StrippedHistoryEvents) GetEvents() []*StrippedHistoryEvent {
if x != nil {
return x.Events
}
return nil
}
// api-linter: core::0158::response-next-page-token-field=disabled
// api-linter: core::0158::response-repeated-first-field=disabled
// api-linter: core::0158::response-plural-first-field=disabled
// GetWorkflowExecutionHistoryResponse is added to the response of GetWorkflowExecutionHistory call. It contains raw
// bytes for history events. The frontend service will deserialize this message as
// temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse because the two message definitions are byte-compatible.
type GetWorkflowExecutionHistoryResponse struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
History [][]byte `protobuf:"bytes,1,rep,name=history,proto3" json:"history,omitempty"`
RawHistory []*v11.DataBlob `protobuf:"bytes,2,rep,name=raw_history,json=rawHistory,proto3" json:"raw_history,omitempty"`
NextPageToken []byte `protobuf:"bytes,3,opt,name=next_page_token,json=nextPageToken,proto3" json:"next_page_token,omitempty"`
Archived bool `protobuf:"varint,4,opt,name=archived,proto3" json:"archived,omitempty"`
}
func (x *GetWorkflowExecutionHistoryResponse) Reset() {
*x = GetWorkflowExecutionHistoryResponse{}
if protoimpl.UnsafeEnabled {
mi := &file_temporal_server_api_history_v1_message_proto_msgTypes[8]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *GetWorkflowExecutionHistoryResponse) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*GetWorkflowExecutionHistoryResponse) ProtoMessage() {}
func (x *GetWorkflowExecutionHistoryResponse) ProtoReflect() protoreflect.Message {
mi := &file_temporal_server_api_history_v1_message_proto_msgTypes[8]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use GetWorkflowExecutionHistoryResponse.ProtoReflect.Descriptor instead.
func (*GetWorkflowExecutionHistoryResponse) Descriptor() ([]byte, []int) {
return file_temporal_server_api_history_v1_message_proto_rawDescGZIP(), []int{8}
}
func (x *GetWorkflowExecutionHistoryResponse) GetHistory() [][]byte {
if x != nil {
return x.History
}
return nil
}
func (x *GetWorkflowExecutionHistoryResponse) GetRawHistory() []*v11.DataBlob {
if x != nil {
return x.RawHistory
}
return nil
}
func (x *GetWorkflowExecutionHistoryResponse) GetNextPageToken() []byte {
if x != nil {
return x.NextPageToken
}
return nil
}
func (x *GetWorkflowExecutionHistoryResponse) GetArchived() bool {
if x != nil {
return x.Archived
}
return false
}
var File_temporal_server_api_history_v1_message_proto protoreflect.FileDescriptor
var file_temporal_server_api_history_v1_message_proto_rawDesc = []byte{
0x0a, 0x2c, 0x74, 0x65, 0x6d, 0x70, 0x6f, 0x72, 0x61, 0x6c, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72,
0x2f, 0x61, 0x70, 0x69, 0x2f, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x2f, 0x76, 0x31, 0x2f, 0x6d,
0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x1e, 0x74, 0x65, 0x6d,
0x70, 0x6f, 0x72, 0x61, 0x6c, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e,
0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x2e, 0x76, 0x31, 0x1a, 0x1f, 0x67, 0x6f, 0x6f, 0x67, 0x6c,
0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73,
0x70, 0x6f, 0x72, 0x61, 0x6c, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69,
0x2e, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x2e, 0x76, 0x31, 0x1a, 0x1f, 0x67, 0x6f, 0x6f, 0x67,
0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73,
0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x25, 0x74, 0x65, 0x6d, 0x70, 0x6f,
0x72, 0x61, 0x6c, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x2f, 0x76,
0x31, 0x2f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x24,
0x74, 0x65, 0x6d, 0x70, 0x6f, 0x72, 0x61, 0x6c, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x63, 0x6f, 0x6d, 0x6d,
0x6f, 0x6e, 0x2f, 0x76, 0x31, 0x2f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x70, 0x72, 0x6f,
0x74, 0x6f, 0x22, 0x79, 0x0a, 0x19, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x69, 0x65, 0x6e, 0x74, 0x57,
0x6f, 0x72, 0x6b, 0x66, 0x6c, 0x6f, 0x77, 0x54, 0x61, 0x73, 0x6b, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x50,
0x0a, 0x0e, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x5f, 0x73, 0x75, 0x66, 0x66, 0x69, 0x78, 0x18,
0x03, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x25, 0x2e, 0x74, 0x65, 0x6d, 0x70, 0x6f, 0x72, 0x61, 0x6c, 0x2e,
0x61, 0x70, 0x69, 0x2e, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x2e, 0x76, 0x31, 0x2e, 0x48, 0x69,
0x73, 0x74, 0x6f, 0x72, 0x79, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x52, 0x0d, 0x68, 0x69, 0x73, 0x74, 0x6f,
0x72, 0x79, 0x53, 0x75, 0x66, 0x66, 0x69, 0x78, 0x42, 0x02, 0x68, 0x00, 0x4a, 0x04, 0x08, 0x01,
0x10, 0x02, 0x4a, 0x04, 0x08, 0x02, 0x10, 0x03, 0x22, 0x51, 0x0a, 0x12, 0x56, 0x65, 0x72, 0x73, 0x69,
0x6f, 0x6e, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x49, 0x74, 0x65, 0x6d, 0x12, 0x1d, 0x0a, 0x08,
0x65, 0x76, 0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x07, 0x65,
0x76, 0x65, 0x6e, 0x74, 0x49, 0x64, 0x42, 0x02, 0x68, 0x00, 0x12, 0x1c, 0x0a, 0x07, 0x76, 0x65, 0x72,
0x73, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x03, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69,
0x6f, 0x6e, 0x42, 0x02, 0x68, 0x00, 0x22, 0x85, 0x01, 0x0a, 0x0e, 0x56, 0x65, 0x72, 0x73, 0x69,
0x6f, 0x6e, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x12, 0x25, 0x0a, 0x0c, 0x62, 0x72, 0x61, 0x6e,
0x63, 0x68, 0x5f, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0b, 0x62,
0x72, 0x61, 0x6e, 0x63, 0x68, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x42, 0x02, 0x68, 0x00, 0x12, 0x4c, 0x0a,
0x05, 0x69, 0x74, 0x65, 0x6d, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x32, 0x2e, 0x74, 0x65,
0x6d, 0x70, 0x6f, 0x72, 0x61, 0x6c, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69,
0x2e, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x2e, 0x76, 0x31, 0x2e, 0x56, 0x65, 0x72, 0x73,
0x69, 0x6f, 0x6e, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x49, 0x74, 0x65, 0x6d, 0x52, 0x05, 0x69,
0x74, 0x65, 0x6d, 0x73, 0x42, 0x02, 0x68, 0x00, 0x22, 0xab, 0x01, 0x0a, 0x10, 0x56, 0x65, 0x72, 0x73,
0x69, 0x6f, 0x6e, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x69, 0x65, 0x73, 0x12, 0x45, 0x0a, 0x1d, 0x63,
0x75, 0x72, 0x72, 0x65, 0x6e, 0x74, 0x5f, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x5f, 0x68, 0x69,
0x73, 0x74, 0x6f, 0x72, 0x79, 0x5f, 0x69, 0x6e, 0x64, 0x65, 0x78, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05,
0x52, 0x1a, 0x63, 0x75, 0x72, 0x72, 0x65, 0x6e, 0x74, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e,
0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x49, 0x6e, 0x64, 0x65, 0x78, 0x42, 0x02, 0x68, 0x00, 0x12,
0x50, 0x0a, 0x09, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x69, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28,
0x0b, 0x32, 0x2e, 0x2e, 0x74, 0x65, 0x6d, 0x70, 0x6f, 0x72, 0x61, 0x6c, 0x2e, 0x73, 0x65, 0x72, 0x76,
0x72, 0x61, 0x6c, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x2f,
0x76, 0x31, 0x2f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22,
0x79, 0x0a, 0x19, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x69, 0x65, 0x6e, 0x74, 0x57, 0x6f, 0x72, 0x6b, 0x66,
0x6c, 0x6f, 0x77, 0x54, 0x61, 0x73, 0x6b, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x50, 0x0a, 0x0e, 0x68, 0x69,
0x73, 0x74, 0x6f, 0x72, 0x79, 0x5f, 0x73, 0x75, 0x66, 0x66, 0x69, 0x78, 0x18, 0x03, 0x20, 0x03,
0x28, 0x0b, 0x32, 0x25, 0x2e, 0x74, 0x65, 0x6d, 0x70, 0x6f, 0x72, 0x61, 0x6c, 0x2e, 0x61, 0x70, 0x69,
0x2e, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x2e, 0x76, 0x31, 0x2e, 0x48, 0x69, 0x73, 0x74, 0x6f,
0x72, 0x79, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x52, 0x0d, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79,
0x53, 0x75, 0x66, 0x66, 0x69, 0x78, 0x42, 0x02, 0x68, 0x00, 0x4a, 0x04, 0x08, 0x01, 0x10, 0x02, 0x4a,
0x04, 0x08, 0x02, 0x10, 0x03, 0x22, 0x51, 0x0a, 0x12, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x48,
0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x49, 0x74, 0x65, 0x6d, 0x12, 0x1d, 0x0a, 0x08, 0x65, 0x76, 0x65,
0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x07, 0x65, 0x76, 0x65,
0x6e, 0x74, 0x49, 0x64, 0x42, 0x02, 0x68, 0x00, 0x12, 0x1c, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69,
0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x03, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e,
0x42, 0x02, 0x68, 0x00, 0x22, 0x85, 0x01, 0x0a, 0x0e, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x48,
0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x12, 0x25, 0x0a, 0x0c, 0x62, 0x72, 0x61, 0x6e, 0x63, 0x68,
0x5f, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0b, 0x62, 0x72, 0x61,
0x6e, 0x63, 0x68, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x42, 0x02, 0x68, 0x00, 0x12, 0x4c, 0x0a, 0x05, 0x69,
0x74, 0x65, 0x6d, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x32, 0x2e, 0x74, 0x65, 0x6d,
0x70, 0x6f, 0x72, 0x61, 0x6c, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e,
0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x2e, 0x76, 0x31, 0x2e, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f,
0x6e, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x49, 0x74, 0x65, 0x6d, 0x52, 0x05, 0x69, 0x74, 0x65,
0x6d, 0x73, 0x42, 0x02, 0x68, 0x00, 0x22, 0xab, 0x01, 0x0a, 0x10, 0x56, 0x65, 0x72, 0x73, 0x69,
0x6f, 0x6e, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x69, 0x65, 0x73, 0x12, 0x45, 0x0a, 0x1d, 0x63, 0x75,
0x72, 0x72, 0x65, 0x6e, 0x74, 0x5f, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x5f, 0x68, 0x69, 0x73,
0x74, 0x6f, 0x72, 0x79, 0x5f, 0x69, 0x6e, 0x64, 0x65, 0x78, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52,
0x1a, 0x63, 0x75, 0x72, 0x72, 0x65, 0x6e, 0x74, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x48,
0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x49, 0x6e, 0x64, 0x65, 0x78, 0x42, 0x02, 0x68, 0x00, 0x12, 0x50,
0x0a, 0x09, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x69, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b,
0x32, 0x2e, 0x2e, 0x74, 0x65, 0x6d, 0x70, 0x6f, 0x72, 0x61, 0x6c, 0x2e, 0x73, 0x65, 0x72, 0x76,
0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x2e, 0x76, 0x31,
0x2e, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x52, 0x09,
0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x69, 0x65, 0x73, 0x42, 0x02, 0x68, 0x00, 0x22, 0x63, 0x0a,
0x07, 0x54, 0x61, 0x73, 0x6b, 0x4b, 0x65, 0x79, 0x12, 0x1b, 0x0a, 0x07, 0x74, 0x61, 0x73, 0x6b, 0x5f,
0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x69, 0x65, 0x73, 0x42, 0x02, 0x68, 0x00, 0x22, 0x63, 0x0a, 0x07,
0x54, 0x61, 0x73, 0x6b, 0x4b, 0x65, 0x79, 0x12, 0x1b, 0x0a, 0x07, 0x74, 0x61, 0x73, 0x6b, 0x5f,
0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x06, 0x74, 0x61, 0x73, 0x6b, 0x49, 0x64, 0x42,
0x02, 0x68, 0x00, 0x12, 0x3b, 0x0a, 0x09, 0x66, 0x69, 0x72, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18,
0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72,
0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52,
0x08, 0x66, 0x69, 0x72, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x42, 0x02, 0x68, 0x00, 0x22, 0xcf, 0x01,
0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70,
0x52, 0x08, 0x66, 0x69, 0x72, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x42, 0x02, 0x68, 0x00, 0x22, 0xcf, 0x01,
0x0a, 0x09, 0x54, 0x61, 0x73, 0x6b, 0x52, 0x61, 0x6e, 0x67, 0x65, 0x12, 0x60, 0x0a, 0x16, 0x69, 0x6e,
0x63, 0x6c, 0x75, 0x73, 0x69, 0x76, 0x65, 0x5f, 0x6d, 0x69, 0x6e, 0x5f, 0x74, 0x61, 0x73, 0x6b, 0x5f,
0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x27, 0x2e, 0x74, 0x65, 0x6d, 0x70, 0x6f,
0x72, 0x61, 0x6c, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x68, 0x69,
0x73, 0x74, 0x6f, 0x72, 0x79, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x61, 0x73, 0x6b, 0x4b, 0x65, 0x79, 0x52,
0x13, 0x69, 0x6e, 0x63, 0x6c, 0x75, 0x73, 0x69, 0x76, 0x65, 0x4d, 0x69, 0x6e, 0x54, 0x61, 0x73,
0x6b, 0x4b, 0x65, 0x79, 0x42, 0x02, 0x68, 0x00, 0x12, 0x60, 0x0a, 0x16, 0x65, 0x78, 0x63, 0x6c, 0x75,
0x73, 0x69, 0x76, 0x65, 0x5f, 0x6d, 0x61, 0x78, 0x5f, 0x74, 0x61, 0x73, 0x6b, 0x5f, 0x6b, 0x65, 0x79,
0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x27, 0x2e, 0x74, 0x65, 0x6d, 0x70, 0x6f, 0x72, 0x61, 0x6c,
0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x68, 0x69, 0x73, 0x74, 0x6f,
0x72, 0x79, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x61, 0x73, 0x6b, 0x4b, 0x65, 0x79, 0x52, 0x13, 0x65, 0x78,
0x63, 0x6c, 0x75, 0x73, 0x69, 0x76, 0x65, 0x4d, 0x61, 0x78, 0x54, 0x61, 0x73, 0x6b, 0x4b, 0x65,
0x79, 0x42, 0x02, 0x68, 0x00, 0x22, 0x53, 0x0a, 0x14, 0x53, 0x74, 0x72, 0x69, 0x70, 0x70, 0x65, 0x64,
0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x12, 0x1d, 0x0a, 0x08, 0x65,
0x76, 0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x07, 0x65, 0x76,
0x65, 0x6e, 0x74, 0x49, 0x64, 0x42, 0x02, 0x68, 0x00, 0x12, 0x1c, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73,
0x69, 0x6f, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x03, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69,
0x6f, 0x6e, 0x42, 0x02, 0x68, 0x00, 0x22, 0x69, 0x0a, 0x15, 0x53, 0x74, 0x72, 0x69, 0x70, 0x70, 0x65,
0x64, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x12, 0x50, 0x0a,
0x06, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x34, 0x2e, 0x74,
0x65, 0x6d, 0x70, 0x6f, 0x72, 0x61, 0x6c, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2e, 0x61, 0x70,
0x69, 0x2e, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x74, 0x72, 0x69,
0x70, 0x70, 0x65, 0x64, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x45, 0x76, 0x65, 0x6e, 0x74,
0x52, 0x06, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x42, 0x02, 0x68, 0x00, 0x22, 0xd6, 0x01, 0x0a, 0x23,
0x47, 0x65, 0x74, 0x57, 0x6f, 0x72, 0x6b, 0x66, 0x6c, 0x6f, 0x77, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74,
0x69, 0x6f, 0x6e, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
0x65, 0x12, 0x1c, 0x0a, 0x07, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x18, 0x01, 0x20, 0x03, 0x28,
0x0c, 0x52, 0x07, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x42, 0x02, 0x68, 0x00, 0x12, 0x45, 0x0a,
0x0b, 0x72, 0x61, 0x77, 0x5f, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x18, 0x02, 0x20, 0x03,
0x28, 0x0b, 0x32, 0x20, 0x2e, 0x74, 0x65, 0x6d, 0x70, 0x6f, 0x72, 0x61, 0x6c, 0x2e, 0x61, 0x70, 0x69,
0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x6f, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x44, 0x61, 0x74, 0x61, 0x42, 0x6c,
0x6f, 0x62, 0x52, 0x0a, 0x72, 0x61, 0x77, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x42, 0x02, 0x68,
0x00, 0x12, 0x2a, 0x0a, 0x0f, 0x6e, 0x65, 0x78, 0x74, 0x5f, 0x70, 0x61, 0x67, 0x65, 0x5f, 0x74, 0x6f,
0x6b, 0x65, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0d, 0x6e, 0x65, 0x78, 0x74, 0x50, 0x61,
0x67, 0x65, 0x54, 0x6f, 0x6b, 0x65, 0x6e, 0x42, 0x02, 0x68, 0x00, 0x12, 0x1e, 0x0a, 0x08, 0x61,
0x72, 0x63, 0x68, 0x69, 0x76, 0x65, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x08, 0x52, 0x08, 0x61, 0x72,
0x63, 0x68, 0x69, 0x76, 0x65, 0x64, 0x42, 0x02, 0x68, 0x00, 0x42, 0x2e, 0x5a, 0x2c, 0x67, 0x6f, 0x2e,
0x74, 0x65, 0x6d, 0x70, 0x6f, 0x72, 0x61, 0x6c, 0x2e, 0x69, 0x6f, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x65,
0x72, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x2f, 0x76, 0x31, 0x3b,
0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x27, 0x2e, 0x74, 0x65, 0x6d, 0x70,
0x6f, 0x72, 0x61, 0x6c, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x68,
0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x61, 0x73, 0x6b, 0x4b, 0x65, 0x79,
0x52, 0x13, 0x69, 0x6e, 0x63, 0x6c, 0x75, 0x73, 0x69, 0x76, 0x65, 0x4d, 0x69, 0x6e, 0x54, 0x61,
0x73, 0x6b, 0x4b, 0x65, 0x79, 0x42, 0x02, 0x68, 0x00, 0x12, 0x60, 0x0a, 0x16, 0x65, 0x78, 0x63, 0x6c,
0x75, 0x73, 0x69, 0x76, 0x65, 0x5f, 0x6d, 0x61, 0x78, 0x5f, 0x74, 0x61, 0x73, 0x6b, 0x5f, 0x6b, 0x65,
0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x27, 0x2e, 0x74, 0x65, 0x6d, 0x70, 0x6f, 0x72, 0x61,
0x6c, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x68, 0x69, 0x73,
0x74, 0x6f, 0x72, 0x79, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x61, 0x73, 0x6b, 0x4b, 0x65, 0x79, 0x52, 0x13,
0x65, 0x78, 0x63, 0x6c, 0x75, 0x73, 0x69, 0x76, 0x65, 0x4d, 0x61, 0x78, 0x54, 0x61, 0x73, 0x6b, 0x4b,
0x65, 0x79, 0x42, 0x02, 0x68, 0x00, 0x42, 0x2e, 0x5a, 0x2c, 0x67, 0x6f, 0x2e, 0x74, 0x65, 0x6d, 0x70,
0x6f, 0x72, 0x61, 0x6c, 0x2e, 0x69, 0x6f, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2f, 0x61,
0x70, 0x69, 0x2f, 0x68, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x2f, 0x76, 0x31, 0x3b, 0x68, 0x69, 0x73,
0x74, 0x6f, 0x72, 0x79, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
@@ -657,35 +450,29 @@ func file_temporal_server_api_history_v1_message_proto_rawDescGZIP() []byte {
return file_temporal_server_api_history_v1_message_proto_rawDescData
}
var file_temporal_server_api_history_v1_message_proto_msgTypes = make([]protoimpl.MessageInfo, 9)
var file_temporal_server_api_history_v1_message_proto_msgTypes = make([]protoimpl.MessageInfo, 6)
var file_temporal_server_api_history_v1_message_proto_goTypes = []interface{}{
(*TransientWorkflowTaskInfo)(nil), // 0: temporal.server.api.history.v1.TransientWorkflowTaskInfo
(*VersionHistoryItem)(nil), // 1: temporal.server.api.history.v1.VersionHistoryItem
(*VersionHistory)(nil), // 2: temporal.server.api.history.v1.VersionHistory
(*VersionHistories)(nil), // 3: temporal.server.api.history.v1.VersionHistories
(*TaskKey)(nil), // 4: temporal.server.api.history.v1.TaskKey
(*TaskRange)(nil), // 5: temporal.server.api.history.v1.TaskRange
(*StrippedHistoryEvent)(nil), // 6: temporal.server.api.history.v1.StrippedHistoryEvent
(*StrippedHistoryEvents)(nil), // 7: temporal.server.api.history.v1.StrippedHistoryEvents
(*GetWorkflowExecutionHistoryResponse)(nil), // 8: temporal.server.api.history.v1.GetWorkflowExecutionHistoryResponse
(*v1.HistoryEvent)(nil), // 9: temporal.api.history.v1.HistoryEvent
(*timestamppb.Timestamp)(nil), // 10: google.protobuf.Timestamp
(*v11.DataBlob)(nil), // 11: temporal.api.common.v1.DataBlob
(*TransientWorkflowTaskInfo)(nil), // 0: temporal.server.api.history.v1.TransientWorkflowTaskInfo
(*VersionHistoryItem)(nil), // 1: temporal.server.api.history.v1.VersionHistoryItem
(*VersionHistory)(nil), // 2: temporal.server.api.history.v1.VersionHistory
(*VersionHistories)(nil), // 3: temporal.server.api.history.v1.VersionHistories
(*TaskKey)(nil), // 4: temporal.server.api.history.v1.TaskKey
(*TaskRange)(nil), // 5: temporal.server.api.history.v1.TaskRange
(*v1.HistoryEvent)(nil), // 6: temporal.api.history.v1.HistoryEvent
(*timestamppb.Timestamp)(nil), // 7: google.protobuf.Timestamp
}
var file_temporal_server_api_history_v1_message_proto_depIdxs = []int32{
9, // 0: temporal.server.api.history.v1.TransientWorkflowTaskInfo.history_suffix:type_name -> temporal.api.history.v1.HistoryEvent
1, // 1: temporal.server.api.history.v1.VersionHistory.items:type_name -> temporal.server.api.history.v1.VersionHistoryItem
2, // 2: temporal.server.api.history.v1.VersionHistories.histories:type_name -> temporal.server.api.history.v1.VersionHistory
10, // 3: temporal.server.api.history.v1.TaskKey.fire_time:type_name -> google.protobuf.Timestamp
4, // 4: temporal.server.api.history.v1.TaskRange.inclusive_min_task_key:type_name -> temporal.server.api.history.v1.TaskKey
4, // 5: temporal.server.api.history.v1.TaskRange.exclusive_max_task_key:type_name -> temporal.server.api.history.v1.TaskKey
6, // 6: temporal.server.api.history.v1.StrippedHistoryEvents.events:type_name -> temporal.server.api.history.v1.StrippedHistoryEvent
11, // 7: temporal.server.api.history.v1.GetWorkflowExecutionHistoryResponse.raw_history:type_name -> temporal.api.common.v1.DataBlob
8, // [8:8] is the sub-list for method output_type
8, // [8:8] is the sub-list for method input_type
8, // [8:8] is the sub-list for extension type_name
8, // [8:8] is the sub-list for extension extendee
0, // [0:8] is the sub-list for field type_name
6, // 0: temporal.server.api.history.v1.TransientWorkflowTaskInfo.history_suffix:type_name -> temporal.api.history.v1.HistoryEvent
1, // 1: temporal.server.api.history.v1.VersionHistory.items:type_name -> temporal.server.api.history.v1.VersionHistoryItem
2, // 2: temporal.server.api.history.v1.VersionHistories.histories:type_name -> temporal.server.api.history.v1.VersionHistory
7, // 3: temporal.server.api.history.v1.TaskKey.fire_time:type_name -> google.protobuf.Timestamp
4, // 4: temporal.server.api.history.v1.TaskRange.inclusive_min_task_key:type_name -> temporal.server.api.history.v1.TaskKey
4, // 5: temporal.server.api.history.v1.TaskRange.exclusive_max_task_key:type_name -> temporal.server.api.history.v1.TaskKey
6, // [6:6] is the sub-list for method output_type
6, // [6:6] is the sub-list for method input_type
6, // [6:6] is the sub-list for extension type_name
6, // [6:6] is the sub-list for extension extendee
0, // [0:6] is the sub-list for field type_name
}
func init() { file_temporal_server_api_history_v1_message_proto_init() }
@@ -766,42 +553,6 @@ func file_temporal_server_api_history_v1_message_proto_init() {
return nil
}
}
file_temporal_server_api_history_v1_message_proto_msgTypes[6].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*StrippedHistoryEvent); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
file_temporal_server_api_history_v1_message_proto_msgTypes[7].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*StrippedHistoryEvents); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
file_temporal_server_api_history_v1_message_proto_msgTypes[8].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*GetWorkflowExecutionHistoryResponse); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
}
type x struct{}
out := protoimpl.TypeBuilder{
@@ -809,7 +560,7 @@ func file_temporal_server_api_history_v1_message_proto_init() {
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: file_temporal_server_api_history_v1_message_proto_rawDesc,
NumEnums: 0,
NumMessages: 9,
NumMessages: 6,
NumExtensions: 0,
NumServices: 0,
},

37
api/historyservice/v1/request_response.go-helpers.pb.go
View File

@@ -4173,43 +4173,6 @@ func (this *GetWorkflowExecutionHistoryResponse) Equal(that interface{}) bool {
return proto.Equal(this, that1)
}
// Marshal an object of type GetWorkflowExecutionHistoryResponseWithRaw to the protobuf v3 wire format
func (val *GetWorkflowExecutionHistoryResponseWithRaw) Marshal() ([]byte, error) {
return proto.Marshal(val)
}
// Unmarshal an object of type GetWorkflowExecutionHistoryResponseWithRaw from the protobuf v3 wire format
func (val *GetWorkflowExecutionHistoryResponseWithRaw) Unmarshal(buf []byte) error {
return proto.Unmarshal(buf, val)
}
// Size returns the size of the object, in bytes, once serialized
func (val *GetWorkflowExecutionHistoryResponseWithRaw) Size() int {
return proto.Size(val)
}
// Equal returns whether two GetWorkflowExecutionHistoryResponseWithRaw values are equivalent by recursively
// comparing the message's fields.
// For more information see the documentation for
// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
func (this *GetWorkflowExecutionHistoryResponseWithRaw) Equal(that interface{}) bool {
if that == nil {
return this == nil
}
var that1 *GetWorkflowExecutionHistoryResponseWithRaw
switch t := that.(type) {
case *GetWorkflowExecutionHistoryResponseWithRaw:
that1 = t
case GetWorkflowExecutionHistoryResponseWithRaw:
that1 = &t
default:
return false
}
return proto.Equal(this, that1)
}
// Marshal an object of type GetWorkflowExecutionHistoryReverseRequest to the protobuf v3 wire format
func (val *GetWorkflowExecutionHistoryReverseRequest) Marshal() ([]byte, error) {
return proto.Marshal(val)

4900
api/historyservice/v1/request_response.pb.go
View File

File diff suppressed because it is too large Load Diff

4
api/historyservice/v1/service_grpc.pb.go
View File

@@ -1254,7 +1254,7 @@ type HistoryServiceServer interface {
// aip.dev/not-precedent: This service does not follow the update method API --)
PollWorkflowExecutionUpdate(context.Context, *PollWorkflowExecutionUpdateRequest) (*PollWorkflowExecutionUpdateResponse, error)
StreamWorkflowReplicationMessages(HistoryService_StreamWorkflowReplicationMessagesServer) error
GetWorkflowExecutionHistory(context.Context, *GetWorkflowExecutionHistoryRequest) (*GetWorkflowExecutionHistoryResponseWithRaw, error)
GetWorkflowExecutionHistory(context.Context, *GetWorkflowExecutionHistoryRequest) (*GetWorkflowExecutionHistoryResponse, error)
GetWorkflowExecutionHistoryReverse(context.Context, *GetWorkflowExecutionHistoryReverseRequest) (*GetWorkflowExecutionHistoryReverseResponse, error)
GetWorkflowExecutionRawHistoryV2(context.Context, *GetWorkflowExecutionRawHistoryV2Request) (*GetWorkflowExecutionRawHistoryV2Response, error)
GetWorkflowExecutionRawHistory(context.Context, *GetWorkflowExecutionRawHistoryRequest) (*GetWorkflowExecutionRawHistoryResponse, error)
@@ -1504,7 +1504,7 @@ func (UnimplementedHistoryServiceServer) PollWorkflowExecutionUpdate(context.Con
func (UnimplementedHistoryServiceServer) StreamWorkflowReplicationMessages(HistoryService_StreamWorkflowReplicationMessagesServer) error {
return status.Errorf(codes.Unimplemented, "method StreamWorkflowReplicationMessages not implemented")
}
func (UnimplementedHistoryServiceServer) GetWorkflowExecutionHistory(context.Context, *GetWorkflowExecutionHistoryRequest) (*GetWorkflowExecutionHistoryResponseWithRaw, error) {
func (UnimplementedHistoryServiceServer) GetWorkflowExecutionHistory(context.Context, *GetWorkflowExecutionHistoryRequest) (*GetWorkflowExecutionHistoryResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method GetWorkflowExecutionHistory not implemented")
}
func (UnimplementedHistoryServiceServer) GetWorkflowExecutionHistoryReverse(context.Context, *GetWorkflowExecutionHistoryReverseRequest) (*GetWorkflowExecutionHistoryReverseResponse, error) {

1
common/persistence/history_manager.go
View File

@@ -824,6 +824,7 @@ func (m *executionManagerImpl) readRawHistoryBranchAndFilter(
token.LastNodeID = lastNode.NodeID
token.LastTransactionID = lastNode.TransactionID
}
return dataBlobs, transactionIDs, nodeIDs, token, dataSize, nil
}

124
common/persistence/history_manager_util.go
View File

@@ -26,16 +26,10 @@ package persistence
import (
"context"
"fmt"
"sort"
commonpb "go.temporal.io/api/common/v1"
historypb "go.temporal.io/api/history/v1"
"go.temporal.io/api/serviceerror"
historyspb "go.temporal.io/server/api/history/v1"
persistencespb "go.temporal.io/server/api/persistence/v1"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/log/tag"
)
// ReadFullPageEvents reads a full page of history events from ExecutionManager. Due to storage format of V2 History
@@ -62,30 +56,6 @@ func ReadFullPageEvents(
}
}
// ReadFullPageRawEvents reads a full page of raw history events from ExecutionManager. Due to storage format of V2 History
// it is not guaranteed that pageSize amount of data is returned. Function returns the list of history blobs, the size
// of data read, the next page token, and an error if present.
func ReadFullPageRawEvents(
ctx context.Context,
executionMgr ExecutionManager,
req *ReadHistoryBranchRequest,
) ([]*commonpb.DataBlob, int, []byte, error) {
var blobs []*commonpb.DataBlob
size := 0
for {
response, err := executionMgr.ReadRawHistoryBranch(ctx, req)
if err != nil {
return nil, 0, nil, err
}
blobs = append(blobs, response.HistoryEventBlobs...)
size += response.Size
if len(blobs) >= req.PageSize || len(response.NextPageToken) == 0 {
return blobs, size, response.NextPageToken, nil
}
req.NextPageToken = response.NextPageToken
}
}
// ReadFullPageEventsByBatch reads a full page of history events by batch from ExecutionManager. Due to storage format of V2 History
// it is not guaranteed that pageSize amount of data is returned. Function returns the list of history batches, the size
// of data read, the next page token, and an error if present.
@@ -158,97 +128,3 @@ func sortAncestors(ans []*persistencespb.HistoryBranchRange) {
}
}
}
func ValidateBatch(
batch []*historyspb.StrippedHistoryEvent,
branchToken []byte,
lastEventID int64,
logger log.Logger,
) error {
var firstEvent, lastEvent *historyspb.StrippedHistoryEvent
var eventCount int
dataLossTags := func(cause string) []tag.Tag {
return []tag.Tag{
tag.Cause(cause),
tag.WorkflowBranchToken(branchToken),
tag.WorkflowFirstEventID(firstEvent.GetEventId()),
tag.FirstEventVersion(firstEvent.GetVersion()),
tag.WorkflowNextEventID(lastEvent.GetEventId()),
tag.LastEventVersion(lastEvent.GetVersion()),
tag.Counter(eventCount),
tag.TokenLastEventID(lastEventID),
}
}
firstEvent = batch[0]
eventCount = len(batch)
lastEvent = batch[eventCount-1]
if firstEvent.GetVersion() != lastEvent.GetVersion() || firstEvent.GetEventId()+int64(eventCount-1) != lastEvent.GetEventId() {
// in a single batch, version should be the same, and ID should be contiguous
logger.Error(dataLossMsg, dataLossTags(errWrongVersion)...)
return serviceerror.NewDataLoss(errWrongVersion)
}
// If it is the first batch in the response, we cannot check the first event id here. That information is in the historyPagingToken.
// TODO: PPV refactor to move this check to ExecutionManager so that we can include that check as well.
if lastEventID != 0 && firstEvent.GetEventId() != lastEventID+1 {
logger.Error(dataLossMsg, dataLossTags(errNonContiguousEventID)...)
return serviceerror.NewDataLoss(errNonContiguousEventID)
}
return nil
}
func VerifyHistoryIsComplete(
events []*historyspb.StrippedHistoryEvent,
expectedFirstEventID int64,
expectedLastEventID int64,
isFirstPage bool,
isLastPage bool,
pageSize int,
) error {
nEvents := len(events)
if nEvents == 0 {
if isLastPage {
// we seem to be returning a non-nil pageToken on the lastPage which
// in turn cases the client to call getHistory again - only to find
// there are no more events to consume - bail out if this is the case here
return nil
}
return serviceerror.NewDataLoss("History contains zero events.")
}
firstEventID := events[0].GetEventId()
lastEventID := events[nEvents-1].GetEventId()
if !isFirstPage { // at least one page of history has been read previously
if firstEventID <= expectedFirstEventID {
// not first page and no events have been read in the previous pages - not possible
return serviceerror.NewDataLoss(fmt.Sprintf("Invalid history: expected first eventID to be > %v but got %v", expectedFirstEventID, firstEventID))
}
expectedFirstEventID = firstEventID
}
if !isLastPage {
// estimate lastEventID based on pageSize. This is a lower bound
// since the persistence layer counts "batch of events" as a single page
expectedLastEventID = expectedFirstEventID + int64(pageSize) - 1
}
nExpectedEvents := expectedLastEventID - expectedFirstEventID + 1
if firstEventID == expectedFirstEventID &&
((isLastPage && lastEventID == expectedLastEventID && int64(nEvents) == nExpectedEvents) ||
(!isLastPage && lastEventID >= expectedLastEventID && int64(nEvents) >= nExpectedEvents)) {
return nil
}
return serviceerror.NewDataLoss(fmt.Sprintf("Incomplete history: expected events [%v-%v] but got events [%v-%v] of length %v: isFirstPage=%v,isLastPage=%v,pageSize=%v",
expectedFirstEventID,
expectedLastEventID,
firstEventID,
lastEventID,
nEvents,
isFirstPage,
isLastPage,
pageSize))
}

29
common/persistence/serialization/serializer.go
View File

@@ -34,7 +34,6 @@ import (
enumspb "go.temporal.io/api/enums/v1"
historypb "go.temporal.io/api/history/v1"
enumsspb "go.temporal.io/server/api/enums/v1"
historyspb "go.temporal.io/server/api/history/v1"
persistencespb "go.temporal.io/server/api/persistence/v1"
replicationspb "go.temporal.io/server/api/replication/v1"
"go.temporal.io/server/common/codec"
@@ -51,7 +50,6 @@ type (
SerializeEvent(event *historypb.HistoryEvent, encodingType enumspb.EncodingType) (*commonpb.DataBlob, error)
DeserializeEvent(data *commonpb.DataBlob) (*historypb.HistoryEvent, error)
DeserializeStrippedEvents(data *commonpb.DataBlob) ([]*historyspb.StrippedHistoryEvent, error)
SerializeClusterMetadata(icm *persistencespb.ClusterMetadata, encodingType enumspb.EncodingType) (*commonpb.DataBlob, error)
DeserializeClusterMetadata(data *commonpb.DataBlob) (*persistencespb.ClusterMetadata, error)
@@ -179,33 +177,6 @@ func (t *serializerImpl) DeserializeEvents(data *commonpb.DataBlob) ([]*historyp
return events.Events, nil
}
func (t *serializerImpl) DeserializeStrippedEvents(data *commonpb.DataBlob) ([]*historyspb.StrippedHistoryEvent, error) {
if data == nil {
return nil, nil
}
if len(data.Data) == 0 {
return nil, nil
}
events := &historyspb.StrippedHistoryEvents{}
var err error
//nolint:exhaustive
switch data.EncodingType {
case enumspb.ENCODING_TYPE_PROTO3:
// Discard unknown fields to improve performance. StrippedHistoryEvents is usually deserialized from HistoryEvent
// which has extra fields that are not needed for this message.
err = proto.UnmarshalOptions{
DiscardUnknown: true,
}.Unmarshal(data.Data, events)
default:
return nil, NewUnknownEncodingTypeError(data.EncodingType.String(), enumspb.ENCODING_TYPE_PROTO3)
}
if err != nil {
return nil, NewDeserializationError(enumspb.ENCODING_TYPE_PROTO3, err)
}
return events.Events, nil
}
func (t *serializerImpl) SerializeEvent(event *historypb.HistoryEvent, encodingType enumspb.EncodingType) (*commonpb.DataBlob, error) {
if event == nil {
return nil, nil

80
common/persistence/serialization/serializer_test.go
View File

@@ -32,7 +32,6 @@ import (
"github.com/stretchr/testify/require"
"github.com/stretchr/testify/suite"
commonpb "go.temporal.io/api/common/v1"
enumspb "go.temporal.io/api/enums/v1"
historypb "go.temporal.io/api/history/v1"
persistencespb "go.temporal.io/server/api/persistence/v1"
@@ -197,85 +196,6 @@ func (s *temporalSerializerSuite) TestSerializeShardInfo_Random() {
s.ProtoEqual(&shardInfo, deserializedShardInfo)
}
func (s *temporalSerializerSuite) TestDeserializeStrippedEvents() {
// 1. Nil data blob
s.Run("NilDataBlob", func() {
events, err := s.serializer.DeserializeStrippedEvents(nil)
s.NoError(err)
s.Nil(events)
})
// 2. Empty data
s.Run("EmptyDataBlob", func() {
// Data is nil
events, err := s.serializer.DeserializeStrippedEvents(&commonpb.DataBlob{
EncodingType: enumspb.ENCODING_TYPE_PROTO3,
Data: nil,
})
s.NoError(err)
s.Nil(events)
// Data is empty byte array
events, err = s.serializer.DeserializeStrippedEvents(&commonpb.DataBlob{
EncodingType: enumspb.ENCODING_TYPE_PROTO3,
Data: []byte{},
})
s.NoError(err)
s.Nil(events)
})
// 3. Unknown encoding type
s.Run("UnknownEncodingType", func() {
_, err := s.serializer.DeserializeStrippedEvents(&commonpb.DataBlob{
EncodingType: enumspb.ENCODING_TYPE_JSON, // Not handled by our switch
Data: []byte("irrelevant-data"),
})
s.Error(err)
s.Contains(err.Error(), "unknown or unsupported encoding type")
})
// 4. Proper proto decoding, discarding unknown fields
s.Run("ProtoDiscardUnknownFields", func() {
// Build a HistoryEvent that contains fields *not* present in StrippedHistoryEvent
historyEvent := &historypb.HistoryEvent{
EventId: 123,
Version: 456,
EventTime: nil, // or a valid timestamp
// This is an extra field not present in StrippedHistoryEvent
Attributes: &historypb.HistoryEvent_WorkflowExecutionStartedEventAttributes{
WorkflowExecutionStartedEventAttributes: &historypb.WorkflowExecutionStartedEventAttributes{
WorkflowType: &commonpb.WorkflowType{Name: "some-workflow-type"},
},
},
}
historyEvents := &historypb.History{
Events: []*historypb.HistoryEvent{historyEvent},
}
// Marshal to protobuf
data, err := historyEvents.Marshal()
s.Require().NoError(err)
dataBlob := &commonpb.DataBlob{
EncodingType: enumspb.ENCODING_TYPE_PROTO3,
Data: data,
}
// Deserialize into StrippedHistoryEvents (should drop unknown fields)
deserializedEvents, err := s.serializer.DeserializeStrippedEvents(dataBlob)
s.NoError(err)
s.Require().Len(deserializedEvents, 1)
// Known fields should be preserved
s.EqualValues(123, deserializedEvents[0].EventId)
s.EqualValues(456, deserializedEvents[0].Version)
reflectMsg := deserializedEvents[0].ProtoReflect()
s.Empty(reflectMsg.GetUnknown(), "Unknown fields should have been discarded")
})
}
func (s *temporalSerializerSuite) TestSerializeWorkflowExecutionState() {
state := &persistencespb.WorkflowExecutionState{
RequestIds: make(map[string]*persistencespb.RequestIDInfo),

2
common/rpc/interceptor/logtags/history_service_server_gen.go
View File

@@ -150,7 +150,7 @@ func (wt *WorkflowTags) extractFromHistoryServiceServerMessage(message any) []ta
tag.WorkflowID(r.GetRequest().GetExecution().GetWorkflowId()),
tag.WorkflowRunID(r.GetRequest().GetExecution().GetRunId()),
}
case *historyservice.GetWorkflowExecutionHistoryResponseWithRaw:
case *historyservice.GetWorkflowExecutionHistoryResponse:
return nil
case *historyservice.GetWorkflowExecutionHistoryReverseRequest:
return []tag.Tag{

5
develop/protoc.sh
View File

@@ -37,11 +37,6 @@ done
color "Update license headers for proto files..."
go run ./cmd/tools/copyright/licensegen.go --scanDir "$new"
color "Modify history service server interface..."
sed -i.bak -e \
's/GetWorkflowExecutionHistory(context\.Context, \*GetWorkflowExecutionHistoryRequest) (\*GetWorkflowExecutionHistoryResponse, error)/GetWorkflowExecutionHistory(context.Context, *GetWorkflowExecutionHistoryRequest) (*GetWorkflowExecutionHistoryResponseWithRaw, error)/g' \
"$new"/temporal/server/api/historyservice/v1/service_grpc.pb.go && rm "$new"/temporal/server/api/historyservice/v1/service_grpc.pb.go.bak
color "Moving proto files into place..."
old=$api.old
[[ -d "$api" ]] && mv -f "$api" "$old"

24
proto/internal/temporal/server/api/history/v1/message.proto
View File

@@ -29,7 +29,6 @@ option go_package = "go.temporal.io/server/api/history/v1;history";
import "google/protobuf/timestamp.proto";
import "temporal/api/history/v1/message.proto";
import "temporal/api/common/v1/message.proto";
message TransientWorkflowTaskInfo {
reserved 1;
@@ -66,26 +65,3 @@ message TaskRange {
TaskKey inclusive_min_task_key = 1;
TaskKey exclusive_max_task_key = 2;
}
// StrippedHistoryEvent is a stripped down version of HistoryEvent that only contains the event_id and version.
message StrippedHistoryEvent {
int64 event_id = 1;
int64 version = 4;
}
message StrippedHistoryEvents {
repeated StrippedHistoryEvent events = 1;
}
// api-linter: core::0158::response-next-page-token-field=disabled
// api-linter: core::0158::response-repeated-first-field=disabled
// api-linter: core::0158::response-plural-first-field=disabled
// GetWorkflowExecutionHistoryResponse is added to the response of GetWorkflowExecutionHistory call. It contains raw
// bytes for history events. The frontend service will deserialize this message as
// temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse because the two message definitions are byte-compatible.
message GetWorkflowExecutionHistoryResponse {
repeated bytes history = 1;
repeated temporal.api.common.v1.DataBlob raw_history = 2;
bytes next_page_token = 3;
bool archived = 4;
}

5
proto/internal/temporal/server/api/historyservice/v1/request_response.proto
View File

@@ -984,11 +984,6 @@ message GetWorkflowExecutionHistoryResponse {
temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse response = 1;
}
// This message must be wire compatible with GetWorkflowExecutionHistoryResponse.
message GetWorkflowExecutionHistoryResponseWithRaw {
temporal.server.api.history.v1.GetWorkflowExecutionHistoryResponse response = 1;
}
message GetWorkflowExecutionHistoryReverseRequest {
option (routing).workflow_id = "request.execution.workflow_id";

1
service/frontend/service.go
View File

@@ -135,6 +135,7 @@ type Config struct {
// VisibilityArchival system protection
VisibilityArchivalQueryMaxPageSize dynamicconfig.IntPropertyFn
// DEPRECATED
SendRawWorkflowHistory dynamicconfig.BoolPropertyFnWithNamespaceFilter
// DefaultWorkflowTaskTimeout the default workflow task timeout

106
service/frontend/workflow_handler.go
View File

@@ -88,7 +88,6 @@ import (
"go.temporal.io/server/common/tqid"
"go.temporal.io/server/common/util"
"go.temporal.io/server/common/worker_versioning"
"go.temporal.io/server/service/history/api"
"go.temporal.io/server/service/worker/batcher"
"go.temporal.io/server/service/worker/deployment"
"go.temporal.io/server/service/worker/scheduler"
@@ -764,23 +763,6 @@ func (wh *WorkflowHandler) GetWorkflowExecutionHistory(ctx context.Context, requ
if err != nil {
return nil, err
}
if !wh.config.SendRawWorkflowHistory(request.Namespace) {
isCloseEventOnly := request.HistoryEventFilterType == enumspb.HISTORY_EVENT_FILTER_TYPE_CLOSE_EVENT
if isCloseEventOnly {
if len(response.Response.History.GetEvents()) > 0 {
response.Response.History.Events = response.Response.History.Events[len(response.Response.History.Events)-1:]
}
}
err = api.ProcessOutgoingSearchAttributes(wh.namespaceRegistry, wh.saProvider, wh.saMapperProvider, response.Response.History.Events, namespaceID, wh.visibilityMgr)
if err != nil {
return nil, err
}
err = fixFollowEvents(ctx, wh.versionChecker, isCloseEventOnly, response.Response.History)
if err != nil {
return nil, err
}
}
return response.Response, nil
}
@@ -5867,91 +5849,3 @@ func (wh *WorkflowHandler) ResetActivity(
return &workflowservice.ResetActivityResponse{}, nil
}
func fixFollowEvents(
ctx context.Context,
versionChecker headers.VersionChecker,
isCloseEventOnly bool,
history *historypb.History,
) error {
// Backwards-compatibility fix for retry events after #1866: older SDKs don't know how to "follow"
// subsequent runs linked in WorkflowExecutionFailed or TimedOut events, so they'll get the wrong result
// when trying to "get" the result of a workflow run. (This applies to cron runs also but "get" on a cron
// workflow isn't really sensible.)
//
// To handle this in a backwards-compatible way, we'll pretend the completion event is actually
// ContinuedAsNew, if it's Failed or TimedOut. We want to do this only when the client is looking for a
// completion event, and not when it's getting the history to display for other purposes. The best signal
// for that purpose is `isCloseEventOnly`. (We can't use `isLongPoll` also because in some cases, older
// versions of the Java SDK don't set that flag.)
//
// TODO: We can remove this once we no longer support SDK versions prior to around September 2021.
// Revisit this once we have an SDK deprecation policy.
followsNextRunId := versionChecker.ClientSupportsFeature(ctx, headers.FeatureFollowsNextRunID)
if isCloseEventOnly && !followsNextRunId && len(history.Events) > 0 {
lastEvent := history.Events[len(history.Events)-1]
fakeEvent, err := makeFakeContinuedAsNewEvent(ctx, lastEvent)
if err != nil {
return err
}
if fakeEvent != nil {
history.Events[len(history.Events)-1] = fakeEvent
}
}
return nil
}
func makeFakeContinuedAsNewEvent(
_ context.Context,
lastEvent *historypb.HistoryEvent,
) (*historypb.HistoryEvent, error) {
//nolint:exhaustive
switch lastEvent.EventType {
case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED:
if lastEvent.GetWorkflowExecutionCompletedEventAttributes().GetNewExecutionRunId() == "" {
return nil, nil
}
case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED:
if lastEvent.GetWorkflowExecutionFailedEventAttributes().GetNewExecutionRunId() == "" {
return nil, nil
}
case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIMED_OUT:
if lastEvent.GetWorkflowExecutionTimedOutEventAttributes().GetNewExecutionRunId() == "" {
return nil, nil
}
default:
return nil, nil
}
// We need to replace the last event with a continued-as-new event that has at least the
// NewExecutionRunId field. We don't actually need any other fields, since that's the only one
// the client looks at in this case, but copy the last result or failure from the real completed
// event just so it's clear what the result was.
newAttrs := &historypb.WorkflowExecutionContinuedAsNewEventAttributes{}
//nolint:exhaustive
switch lastEvent.EventType {
case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED:
attrs := lastEvent.GetWorkflowExecutionCompletedEventAttributes()
newAttrs.NewExecutionRunId = attrs.NewExecutionRunId
newAttrs.LastCompletionResult = attrs.Result
case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED:
attrs := lastEvent.GetWorkflowExecutionFailedEventAttributes()
newAttrs.NewExecutionRunId = attrs.NewExecutionRunId
newAttrs.Failure = attrs.Failure
case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIMED_OUT:
attrs := lastEvent.GetWorkflowExecutionTimedOutEventAttributes()
newAttrs.NewExecutionRunId = attrs.NewExecutionRunId
newAttrs.Failure = failure.NewTimeoutFailure("workflow timeout", enumspb.TIMEOUT_TYPE_START_TO_CLOSE)
}
return &historypb.HistoryEvent{
EventId: lastEvent.EventId,
EventTime: lastEvent.EventTime,
EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW,
Version: lastEvent.Version,
TaskId: lastEvent.TaskId,
Attributes: &historypb.HistoryEvent_WorkflowExecutionContinuedAsNewEventAttributes{
WorkflowExecutionContinuedAsNewEventAttributes: newAttrs,
},
}, nil
}

165
service/frontend/workflow_handler_test.go
View File

@@ -39,7 +39,6 @@ import (
batchpb "go.temporal.io/api/batch/v1"
commonpb "go.temporal.io/api/common/v1"
enumspb "go.temporal.io/api/enums/v1"
failurepb "go.temporal.io/api/failure/v1"
filterpb "go.temporal.io/api/filter/v1"
historypb "go.temporal.io/api/history/v1"
namespacepb "go.temporal.io/api/namespace/v1"
@@ -49,7 +48,6 @@ import (
updatepb "go.temporal.io/api/update/v1"
workflowpb "go.temporal.io/api/workflow/v1"
"go.temporal.io/api/workflowservice/v1"
historyspb "go.temporal.io/server/api/history/v1"
"go.temporal.io/server/api/historyservice/v1"
"go.temporal.io/server/api/historyservicemock/v1"
"go.temporal.io/server/api/matchingservice/v1"
@@ -75,7 +73,7 @@ import (
"go.temporal.io/server/common/rpc/interceptor"
"go.temporal.io/server/common/searchattribute"
"go.temporal.io/server/common/tasktoken"
"go.temporal.io/server/service/history/tests"
e "go.temporal.io/server/service/history/events"
"go.temporal.io/server/service/worker/batcher"
"go.temporal.io/server/service/worker/scheduler"
"go.uber.org/mock/gomock"
@@ -2038,16 +2036,16 @@ func (s *WorkflowHandlerSuite) TestCountWorkflowExecutions() {
}
func (s *WorkflowHandlerSuite) TestVerifyHistoryIsComplete() {
events := make([]*historyspb.StrippedHistoryEvent, 50)
events := make([]*historypb.HistoryEvent, 50)
for i := 0; i < len(events); i++ {
events[i] = &historyspb.StrippedHistoryEvent{EventId: int64(i + 1)}
events[i] = &historypb.HistoryEvent{EventId: int64(i + 1)}
}
var eventsWithHoles []*historyspb.StrippedHistoryEvent
var eventsWithHoles []*historypb.HistoryEvent
eventsWithHoles = append(eventsWithHoles, events[9:12]...)
eventsWithHoles = append(eventsWithHoles, events[20:31]...)
testCases := []struct {
events []*historyspb.StrippedHistoryEvent
events []*historypb.HistoryEvent
firstEventID int64
lastEventID int64
isFirstPage bool
@@ -2081,7 +2079,7 @@ func (s *WorkflowHandlerSuite) TestVerifyHistoryIsComplete() {
}
for i, tc := range testCases {
err := persistence.VerifyHistoryIsComplete(tc.events, tc.firstEventID, tc.lastEventID, tc.isFirstPage, tc.isLastPage, tc.pageSize)
err := e.VerifyHistoryIsComplete(tc.events, tc.firstEventID, tc.lastEventID, tc.isFirstPage, tc.isLastPage, tc.pageSize)
if tc.isResultErr {
s.Error(err, "testcase %v failed", i)
} else {
@@ -2804,78 +2802,6 @@ func (s *WorkflowHandlerSuite) TestListBatchOperations_InvalidRerquest() {
s.ErrorAs(err, &invalidArgumentErr)
}
func (s *WorkflowHandlerSuite) TestGetWorkflowExecutionHistory_FailedConvertedToContinueAsNew() {
config := s.newConfig()
wh := s.getWorkflowHandler(config)
we := commonpb.WorkflowExecution{WorkflowId: "wid1", RunId: uuid.New().String()}
newRunID := uuid.New().String()
s.mockNamespaceCache.EXPECT().GetNamespaceID(tests.Namespace).Return(tests.NamespaceID, nil).AnyTimes()
s.mockNamespaceCache.EXPECT().GetNamespaceName(tests.NamespaceID).Return(tests.Namespace, nil).AnyTimes()
s.mockSearchAttributesProvider.EXPECT().GetSearchAttributes(gomock.Any(), gomock.Any()).Return(searchattribute.TestNameTypeMap, nil).AnyTimes()
req := &workflowservice.GetWorkflowExecutionHistoryRequest{
Namespace: tests.Namespace.String(),
Execution: &we,
MaximumPageSize: 10,
HistoryEventFilterType: enumspb.HISTORY_EVENT_FILTER_TYPE_CLOSE_EVENT,
SkipArchival: true,
}
s.mockHistoryClient.EXPECT().GetWorkflowExecutionHistory(gomock.Any(), &historyservice.GetWorkflowExecutionHistoryRequest{
NamespaceId: tests.NamespaceID.String(),
Request: req,
}).Return(&historyservice.GetWorkflowExecutionHistoryResponse{
Response: &workflowservice.GetWorkflowExecutionHistoryResponse{
History: &historypb.History{
Events: []*historypb.HistoryEvent{
{
EventId: int64(5),
EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED,
Attributes: &historypb.HistoryEvent_WorkflowExecutionFailedEventAttributes{
WorkflowExecutionFailedEventAttributes: &historypb.WorkflowExecutionFailedEventAttributes{
Failure: &failurepb.Failure{Message: "this workflow failed"},
RetryState: enumspb.RETRY_STATE_IN_PROGRESS,
WorkflowTaskCompletedEventId: 4,
NewExecutionRunId: newRunID,
},
},
},
},
},
},
}, nil).Times(2)
oldGoSDKVersion := "1.9.1"
newGoSDKVersion := "1.10.1"
// new sdk: should see failed event
ctx := headers.SetVersionsForTests(context.Background(), newGoSDKVersion, headers.ClientNameGoSDK, headers.SupportedServerVersions, headers.AllFeatures)
resp, err := wh.GetWorkflowExecutionHistory(ctx, req)
s.NoError(err)
s.False(resp.Archived)
event := resp.History.Events[0]
s.Equal(int64(5), event.EventId)
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED, event.EventType)
attrs := event.GetWorkflowExecutionFailedEventAttributes()
s.Equal("this workflow failed", attrs.Failure.Message)
s.Equal(newRunID, attrs.NewExecutionRunId)
s.Equal(enumspb.RETRY_STATE_IN_PROGRESS, attrs.RetryState)
// old sdk: should see continued-as-new event
// TODO: We can remove this once we no longer support SDK versions prior to around September 2021.
// See comment in workflowHandler.go:GetWorkflowExecutionHistory
ctx = headers.SetVersionsForTests(context.Background(), oldGoSDKVersion, headers.ClientNameGoSDK, headers.SupportedServerVersions, "")
resp, err = wh.GetWorkflowExecutionHistory(ctx, req)
s.NoError(err)
s.False(resp.Archived)
event = resp.History.Events[0]
s.Equal(int64(5), event.EventId)
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW, event.EventType)
attrs2 := event.GetWorkflowExecutionContinuedAsNewEventAttributes()
s.Equal(newRunID, attrs2.NewExecutionRunId)
s.Equal("this workflow failed", attrs2.Failure.Message)
}
func (s *WorkflowHandlerSuite) newConfig() *Config {
return NewConfig(dc.NewNoopCollection(), numHistoryShards)
}
@@ -3313,82 +3239,3 @@ func (s *WorkflowHandlerSuite) TestShutdownWorker() {
s.Fail("ShutdownWorker failed:", err)
}
}
// HistoryService returns a different GetWorkflowExecutionHistoryResponse GetWorkflowExecutionHistoryResponseWithRaw to
// WorkflowHandler. Since HistoryClient is defined with the response type GetWorkflowExecutionHistoryResponse, grpc
// will deserialize this message to GetWorkflowExecutionHistoryResponse. This is done to avoid the extra CPU usage in
// history service to deserialize event blobs to []*HistoryEvent. This test ensures that
// GetWorkflowExecutionHistoryResponseWithRaw is correctly deserialized to GetWorkflowExecutionHistoryResponse.
func (s *WorkflowHandlerSuite) TestGetWorkflowExecutionHistoryResponseWithRawHistoryEvents() {
// Create history events and batches
fullHistory := &historypb.History{
Events: []*historypb.HistoryEvent{
{
EventId: 1,
EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED,
Version: 100,
},
{
EventId: 2,
EventType: enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED,
Version: 101,
},
{
EventId: 3,
EventType: enumspb.EVENT_TYPE_ACTIVITY_TASK_COMPLETED,
Version: 102,
},
{
EventId: 4,
EventType: enumspb.EVENT_TYPE_TIMER_FIRED,
Version: 103,
},
},
}
batch1 := &historypb.History{
Events: fullHistory.Events[:2],
}
batch2 := &historypb.History{
Events: fullHistory.Events[2:],
}
// Marshal each batch
rawHistory1, err := batch1.Marshal()
s.Require().NoError(err)
db1 := &commonpb.DataBlob{
EncodingType: enumspb.ENCODING_TYPE_PROTO3,
Data: rawHistory1,
}
rawHistory2, err := batch2.Marshal()
s.Require().NoError(err)
db2 := &commonpb.DataBlob{
EncodingType: enumspb.ENCODING_TYPE_PROTO3,
Data: rawHistory2,
}
// Create a GetWorkflowExecutionHistoryResponse with raw history blobs. This should be wire compatible with
// workflowservice.GetWorkflowExecutionHistoryResponse which has repeated HistoryEvent field instead of repeated bytes.
rawResp := &historyspb.GetWorkflowExecutionHistoryResponse{
History: [][]byte{db1.Data, db2.Data},
}
serializedRawResp, err := rawResp.Marshal()
s.Require().NoError(err)
resp := &workflowservice.GetWorkflowExecutionHistoryResponse{}
err = resp.Unmarshal(serializedRawResp)
s.Require().NoError(err)
// Verify resp has same list of history events as fullHistory
for i, event := range resp.History.Events {
s.Equal(fullHistory.Events[i].EventId, event.EventId)
s.Equal(fullHistory.Events[i].Version, event.Version)
s.Equal(fullHistory.Events[i].EventType, event.EventType)
s.Nil(event.Attributes)
}
}

108
service/history/api/get_history_util.go
View File

@@ -43,6 +43,7 @@ import (
"go.temporal.io/server/common/rpc/interceptor"
"go.temporal.io/server/common/searchattribute"
"go.temporal.io/server/service/history/consts"
"go.temporal.io/server/service/history/events"
"go.temporal.io/server/service/history/shard"
)
@@ -54,61 +55,28 @@ func GetRawHistory(
firstEventID int64,
nextEventID int64,
pageSize int32,
token []byte,
nextPageToken []byte,
transientWorkflowTaskInfo *historyspb.TransientWorkflowTaskInfo,
branchToken []byte,
) ([]*commonpb.DataBlob, []byte, error) {
shardID := common.WorkflowIDToHistoryShard(namespaceID.String(), execution.GetWorkflowId(), shard.GetConfig().NumberOfShards)
logger := shard.GetLogger()
rawHistory, size, nextToken, err := persistence.ReadFullPageRawEvents(
ctx, shard.GetExecutionManager(),
&persistence.ReadHistoryBranchRequest{
BranchToken: branchToken,
MinEventID: firstEventID,
MaxEventID: nextEventID,
PageSize: int(pageSize),
NextPageToken: token,
ShardID: shardID,
},
)
persistenceExecutionManager := shard.GetExecutionManager()
resp, err := persistenceExecutionManager.ReadRawHistoryBranch(ctx, &persistence.ReadHistoryBranchRequest{
BranchToken: branchToken,
MinEventID: firstEventID,
MaxEventID: nextEventID,
PageSize: int(pageSize),
NextPageToken: nextPageToken,
ShardID: shardID,
})
if err != nil {
return nil, nil, err
}
allEvents := make([]*historyspb.StrippedHistoryEvent, 0)
var lastEventID int64
for _, blob := range rawHistory {
events, err := shard.GetPayloadSerializer().DeserializeStrippedEvents(blob)
if err != nil {
return nil, nil, err
}
err = persistence.ValidateBatch(events, branchToken, lastEventID, logger)
if err != nil {
return nil, nil, err
}
allEvents = append(allEvents, events...)
lastEventID = events[len(events)-1].GetEventId()
}
if err = persistence.VerifyHistoryIsComplete(
allEvents,
firstEventID,
nextEventID-1,
len(token) == 0,
len(nextToken) == 0,
int(pageSize),
); err != nil {
metricsHandler := interceptor.GetMetricsHandlerFromContext(ctx, logger).WithTags(metrics.OperationTag(metrics.HistoryGetHistoryScope))
metrics.ServiceErrIncompleteHistoryCounter.With(metricsHandler).Record(1)
logger.Error("getHistory: incomplete history",
tag.WorkflowBranchToken(branchToken),
tag.Error(err))
}
rawHistory := resp.HistoryEventBlobs
metricsHandler := interceptor.GetMetricsHandlerFromContext(ctx, shard.GetLogger()).WithTags(metrics.OperationTag(metrics.HistoryGetHistoryScope))
metrics.HistorySize.With(metricsHandler).Record(int64(size))
if len(nextToken) == 0 && transientWorkflowTaskInfo != nil {
if len(resp.NextPageToken) == 0 && transientWorkflowTaskInfo != nil {
if err := validateTransientWorkflowTaskEvents(nextEventID, transientWorkflowTaskInfo); err != nil {
logger := shard.GetLogger()
metricsHandler := interceptor.GetMetricsHandlerFromContext(ctx, logger).WithTags(metrics.OperationTag(metrics.HistoryGetRawHistoryScope))
@@ -121,15 +89,16 @@ func GetRawHistory(
return nil, nil, err
}
if len(transientWorkflowTaskInfo.HistorySuffix) > 0 {
blob, err := shard.GetPayloadSerializer().SerializeEvents(transientWorkflowTaskInfo.HistorySuffix, enumspb.ENCODING_TYPE_PROTO3)
for _, event := range transientWorkflowTaskInfo.HistorySuffix {
blob, err := shard.GetPayloadSerializer().SerializeEvent(event, enumspb.ENCODING_TYPE_PROTO3)
if err != nil {
return nil, nil, err
}
rawHistory = append(rawHistory, blob)
}
}
return rawHistory, nextToken, nil
return rawHistory, resp.NextPageToken, nil
}
func GetHistory(
@@ -147,6 +116,7 @@ func GetHistory(
) (*historypb.History, []byte, error) {
var size int
isFirstPage := len(nextPageToken) == 0
shardID := common.WorkflowIDToHistoryShard(namespaceID.String(), execution.GetWorkflowId(), shard.GetConfig().NumberOfShards)
var err error
var historyEvents []*historypb.HistoryEvent
@@ -178,6 +148,22 @@ func GetHistory(
metricsHandler := interceptor.GetMetricsHandlerFromContext(ctx, logger).WithTags(metrics.OperationTag(metrics.HistoryGetHistoryScope))
metrics.HistorySize.With(metricsHandler).Record(int64(size))
isLastPage := len(nextPageToken) == 0
if err := events.VerifyHistoryIsComplete(
historyEvents,
firstEventID,
nextEventID-1,
isFirstPage,
isLastPage,
int(pageSize)); err != nil {
metrics.ServiceErrIncompleteHistoryCounter.With(metricsHandler).Record(1)
logger.Error("getHistory: incomplete history",
tag.WorkflowNamespaceID(namespaceID.String()),
tag.WorkflowID(execution.GetWorkflowId()),
tag.WorkflowRunID(execution.GetRunId()),
tag.Error(err))
}
if len(nextPageToken) == 0 && transientWorkflowTaskInfo != nil {
if err := validateTransientWorkflowTaskEvents(nextEventID, transientWorkflowTaskInfo); err != nil {
metrics.ServiceErrIncompleteHistoryCounter.With(metricsHandler).Record(1)
@@ -191,13 +177,7 @@ func GetHistory(
historyEvents = append(historyEvents, transientWorkflowTaskInfo.HistorySuffix...)
}
if err := ProcessOutgoingSearchAttributes(
shard.GetNamespaceRegistry(),
shard.GetSearchAttributesProvider(),
shard.GetSearchAttributesMapperProvider(),
historyEvents,
namespaceID,
persistenceVisibilityMgr); err != nil {
if err := ProcessOutgoingSearchAttributes(shard, historyEvents, namespaceID, persistenceVisibilityMgr); err != nil {
return nil, nil, err
}
@@ -248,13 +228,7 @@ func GetHistoryReverse(
metricsHandler := interceptor.GetMetricsHandlerFromContext(ctx, logger).WithTags(metrics.OperationTag(metrics.HistoryGetHistoryReverseScope))
metrics.HistorySize.With(metricsHandler).Record(int64(size))
if err := ProcessOutgoingSearchAttributes(
shard.GetNamespaceRegistry(),
shard.GetSearchAttributesProvider(),
shard.GetSearchAttributesMapperProvider(),
historyEvents,
namespaceID,
persistenceVisibilityMgr); err != nil {
if err := ProcessOutgoingSearchAttributes(shard, historyEvents, namespaceID, persistenceVisibilityMgr); err != nil {
return nil, nil, 0, err
}
@@ -273,18 +247,16 @@ func GetHistoryReverse(
}
func ProcessOutgoingSearchAttributes(
nsRegistry namespace.Registry,
saProvider searchattribute.Provider,
saMapperProvider searchattribute.MapperProvider,
shardCtx shard.Context,
events []*historypb.HistoryEvent,
namespaceId namespace.ID,
persistenceVisibilityMgr manager.VisibilityManager,
) error {
ns, err := nsRegistry.GetNamespaceName(namespaceId)
ns, err := shardCtx.GetNamespaceRegistry().GetNamespaceName(namespaceId)
if err != nil {
return err
}
saTypeMap, err := saProvider.GetSearchAttributes(persistenceVisibilityMgr.GetIndexName(), false)
saTypeMap, err := shardCtx.GetSearchAttributesProvider().GetSearchAttributes(persistenceVisibilityMgr.GetIndexName(), false)
if err != nil {
return serviceerror.NewUnavailable(fmt.Sprintf(consts.ErrUnableToGetSearchAttributesMessage, err))
}
@@ -302,7 +274,7 @@ func ProcessOutgoingSearchAttributes(
}
if searchAttributes != nil {
searchattribute.ApplyTypeMap(searchAttributes, saTypeMap)
aliasedSas, err := searchattribute.AliasFields(saMapperProvider, searchAttributes, ns.String())
aliasedSas, err := searchattribute.AliasFields(shardCtx.GetSearchAttributesMapperProvider(), searchAttributes, ns.String())
if err != nil {
return err
}

213
service/history/api/getworkflowexecutionhistory/api.go
View File

@@ -31,11 +31,13 @@ import (
enumspb "go.temporal.io/api/enums/v1"
historypb "go.temporal.io/api/history/v1"
"go.temporal.io/api/serviceerror"
"go.temporal.io/api/workflowservice/v1"
historyspb "go.temporal.io/server/api/history/v1"
"go.temporal.io/server/api/historyservice/v1"
persistencespb "go.temporal.io/server/api/persistence/v1"
tokenspb "go.temporal.io/server/api/token/v1"
"go.temporal.io/server/common"
"go.temporal.io/server/common/failure"
"go.temporal.io/server/common/headers"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/common/persistence/transitionhistory"
@@ -55,7 +57,7 @@ func Invoke(
eventNotifier events.Notifier,
request *historyservice.GetWorkflowExecutionHistoryRequest,
persistenceVisibilityMgr manager.VisibilityManager,
) (_ *historyservice.GetWorkflowExecutionHistoryResponseWithRaw, retError error) {
) (_ *historyservice.GetWorkflowExecutionHistoryResponse, retError error) {
namespaceID := namespace.ID(request.GetNamespaceId())
err := api.ValidateNamespaceUUID(namespaceID)
if err != nil {
@@ -191,27 +193,50 @@ func Invoke(
}
}()
history := &historypb.History{}
history.Events = []*historypb.HistoryEvent{}
var historyBlob []*commonpb.DataBlob
if isCloseEventOnly {
if !isWorkflowRunning {
historyBlob, _, err = api.GetRawHistory(
ctx,
shardContext,
namespaceID,
execution,
lastFirstEventID,
nextEventID,
request.Request.GetMaximumPageSize(),
nil,
continuationToken.TransientWorkflowTask,
continuationToken.BranchToken,
)
if err != nil {
return nil, err
}
if shardContext.GetConfig().SendRawWorkflowHistory(request.Request.GetNamespace()) {
historyBlob, _, err = api.GetRawHistory(
ctx,
shardContext,
namespaceID,
execution,
lastFirstEventID,
nextEventID,
request.Request.GetMaximumPageSize(),
nil,
continuationToken.TransientWorkflowTask,
continuationToken.BranchToken,
)
if err != nil {
return nil, err
}
// since getHistory func will not return empty history, so the below is safe
historyBlob = historyBlob[len(historyBlob)-1:]
// since getHistory func will not return empty history, so the below is safe
historyBlob = historyBlob[len(historyBlob)-1:]
} else {
history, _, err = api.GetHistory(
ctx,
shardContext,
namespaceID,
execution,
lastFirstEventID,
nextEventID,
request.Request.GetMaximumPageSize(),
nil,
continuationToken.TransientWorkflowTask,
continuationToken.BranchToken,
persistenceVisibilityMgr,
)
if err != nil {
return nil, err
}
// since getHistory func will not return empty history, so the below is safe
history.Events = history.Events[len(history.Events)-1 : len(history.Events)]
}
continuationToken = nil
} else if isLongPoll {
// set the persistence token to be nil so next time we will query history for updates
@@ -223,22 +248,40 @@ func Invoke(
// return all events
if continuationToken.FirstEventId >= continuationToken.NextEventId {
// currently there is no new event
history.Events = []*historypb.HistoryEvent{}
if !isWorkflowRunning {
continuationToken = nil
}
} else {
historyBlob, continuationToken.PersistenceToken, err = api.GetRawHistory(
ctx,
shardContext,
namespaceID,
execution,
continuationToken.FirstEventId,
continuationToken.NextEventId,
request.Request.GetMaximumPageSize(),
continuationToken.PersistenceToken,
continuationToken.TransientWorkflowTask,
continuationToken.BranchToken,
)
if shardContext.GetConfig().SendRawWorkflowHistory(request.Request.GetNamespace()) {
historyBlob, continuationToken.PersistenceToken, err = api.GetRawHistory(
ctx,
shardContext,
namespaceID,
execution,
continuationToken.FirstEventId,
continuationToken.NextEventId,
request.Request.GetMaximumPageSize(),
continuationToken.PersistenceToken,
continuationToken.TransientWorkflowTask,
continuationToken.BranchToken,
)
} else {
history, continuationToken.PersistenceToken, err = api.GetHistory(
ctx,
shardContext,
namespaceID,
execution,
continuationToken.FirstEventId,
continuationToken.NextEventId,
request.Request.GetMaximumPageSize(),
continuationToken.PersistenceToken,
continuationToken.TransientWorkflowTask,
continuationToken.BranchToken,
persistenceVisibilityMgr,
)
}
if err != nil {
return nil, err
}
@@ -257,32 +300,92 @@ func Invoke(
return nil, err
}
resp := &historyservice.GetWorkflowExecutionHistoryResponseWithRaw{
Response: &historyspb.GetWorkflowExecutionHistoryResponse{
History: nil,
RawHistory: nil,
// Backwards-compatibility fix for retry events after #1866: older SDKs don't know how to "follow"
// subsequent runs linked in WorkflowExecutionFailed or TimedOut events, so they'll get the wrong result
// when trying to "get" the result of a workflow run. (This applies to cron runs also but "get" on a cron
// workflow isn't really sensible.)
//
// To handle this in a backwards-compatible way, we'll pretend the completion event is actually
// ContinuedAsNew, if it's Failed or TimedOut. We want to do this only when the client is looking for a
// completion event, and not when it's getting the history to display for other purposes. The best signal
// for that purpose is `isCloseEventOnly`. (We can't use `isLongPoll` also because in some cases, older
// versions of the Java SDK don't set that flag.)
//
// TODO: We can remove this once we no longer support SDK versions prior to around September 2021.
// Revisit this once we have an SDK deprecation policy.
followsNextRunId := versionChecker.ClientSupportsFeature(ctx, headers.FeatureFollowsNextRunID)
if isCloseEventOnly && !followsNextRunId && len(history.Events) > 0 {
lastEvent := history.Events[len(history.Events)-1]
fakeEvent, err := makeFakeContinuedAsNewEvent(ctx, lastEvent)
if err != nil {
return nil, err
}
if fakeEvent != nil {
history.Events[len(history.Events)-1] = fakeEvent
}
}
return &historyservice.GetWorkflowExecutionHistoryResponse{
Response: &workflowservice.GetWorkflowExecutionHistoryResponse{
History: history,
RawHistory: historyBlob,
NextPageToken: nextToken,
Archived: false,
},
}
if shardContext.GetConfig().SendRawWorkflowHistory(request.Request.GetNamespace()) {
resp.Response.RawHistory = historyBlob
} else {
fullHistory := make([][]byte, 0)
for _, blob := range historyBlob {
fullHistory = append(fullHistory, blob.Data)
}
// If there are no events in the history, frontend will not be able to deserialize the response to History object.
// In that case, create an empty history object and set it in the response.
if len(fullHistory) == 0 {
history := historypb.History{}
blob, err := history.Marshal()
if err != nil {
return nil, err
}
fullHistory = append(fullHistory, blob)
}
resp.Response.History = fullHistory
}
return resp, nil
}, nil
}
func makeFakeContinuedAsNewEvent(
_ context.Context,
lastEvent *historypb.HistoryEvent,
) (*historypb.HistoryEvent, error) {
// nolint:exhaustive
switch lastEvent.EventType {
case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED:
if lastEvent.GetWorkflowExecutionCompletedEventAttributes().GetNewExecutionRunId() == "" {
return nil, nil
}
case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED:
if lastEvent.GetWorkflowExecutionFailedEventAttributes().GetNewExecutionRunId() == "" {
return nil, nil
}
case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIMED_OUT:
if lastEvent.GetWorkflowExecutionTimedOutEventAttributes().GetNewExecutionRunId() == "" {
return nil, nil
}
default:
return nil, nil
}
// We need to replace the last event with a continued-as-new event that has at least the
// NewExecutionRunId field. We don't actually need any other fields, since that's the only one
// the client looks at in this case, but copy the last result or failure from the real completed
// event just so it's clear what the result was.
newAttrs := &historypb.WorkflowExecutionContinuedAsNewEventAttributes{}
// nolint:exhaustive
switch lastEvent.EventType {
case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED:
attrs := lastEvent.GetWorkflowExecutionCompletedEventAttributes()
newAttrs.NewExecutionRunId = attrs.NewExecutionRunId
newAttrs.LastCompletionResult = attrs.Result
case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED:
attrs := lastEvent.GetWorkflowExecutionFailedEventAttributes()
newAttrs.NewExecutionRunId = attrs.NewExecutionRunId
newAttrs.Failure = attrs.Failure
case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIMED_OUT:
attrs := lastEvent.GetWorkflowExecutionTimedOutEventAttributes()
newAttrs.NewExecutionRunId = attrs.NewExecutionRunId
newAttrs.Failure = failure.NewTimeoutFailure("workflow timeout", enumspb.TIMEOUT_TYPE_START_TO_CLOSE)
}
return &historypb.HistoryEvent{
EventId: lastEvent.EventId,
EventTime: lastEvent.EventTime,
EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW,
Version: lastEvent.Version,
TaskId: lastEvent.TaskId,
Attributes: &historypb.HistoryEvent_WorkflowExecutionContinuedAsNewEventAttributes{
WorkflowExecutionContinuedAsNewEventAttributes: newAttrs,
},
}, nil
}

8
service/history/api/startworkflow/api.go
View File

@@ -730,13 +730,7 @@ func (s *Starter) generateResponse(
}, nil
}
if err := api.ProcessOutgoingSearchAttributes(
shardCtx.GetNamespaceRegistry(),
shardCtx.GetSearchAttributesProvider(),
shardCtx.GetSearchAttributesMapperProvider(),
historyEvents,
s.namespace.ID(),
s.visibilityManager); err != nil {
if err := api.ProcessOutgoingSearchAttributes(s.shardContext, historyEvents, s.namespace.ID(), s.visibilityManager); err != nil {
return nil, err
}

88
service/history/events/util.go Normal file
View File

@@ -0,0 +1,88 @@
// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 Uber Technologies, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
package events
import (
"fmt"
historypb "go.temporal.io/api/history/v1"
"go.temporal.io/api/serviceerror"
)
func VerifyHistoryIsComplete(
events []*historypb.HistoryEvent,
expectedFirstEventID int64,
expectedLastEventID int64,
isFirstPage bool,
isLastPage bool,
pageSize int,
) error {
nEvents := len(events)
if nEvents == 0 {
if isLastPage {
// we seem to be returning a non-nil pageToken on the lastPage which
// in turn cases the client to call getHistory again - only to find
// there are no more events to consume - bail out if this is the case here
return nil
}
return serviceerror.NewDataLoss("History contains zero events.")
}
firstEventID := events[0].GetEventId()
lastEventID := events[nEvents-1].GetEventId()
if !isFirstPage { // at least one page of history has been read previously
if firstEventID <= expectedFirstEventID {
// not first page and no events have been read in the previous pages - not possible
return serviceerror.NewDataLoss(fmt.Sprintf("Invalid history: expected first eventID to be > %v but got %v", expectedFirstEventID, firstEventID))
}
expectedFirstEventID = firstEventID
}
if !isLastPage {
// estimate lastEventID based on pageSize. This is a lower bound
// since the persistence layer counts "batch of events" as a single page
expectedLastEventID = expectedFirstEventID + int64(pageSize) - 1
}
nExpectedEvents := expectedLastEventID - expectedFirstEventID + 1
if firstEventID == expectedFirstEventID &&
((isLastPage && lastEventID == expectedLastEventID && int64(nEvents) == nExpectedEvents) ||
(!isLastPage && lastEventID >= expectedLastEventID && int64(nEvents) >= nExpectedEvents)) {
return nil
}
return serviceerror.NewDataLoss(fmt.Sprintf("Incomplete history: expected events [%v-%v] but got events [%v-%v] of length %v: isFirstPage=%v,isLastPage=%v,pageSize=%v",
expectedFirstEventID,
expectedLastEventID,
firstEventID,
lastEventID,
nEvents,
isFirstPage,
isLastPage,
pageSize))
}

2
service/history/handler.go
View File

@@ -2151,7 +2151,7 @@ func (h *Handler) StreamWorkflowReplicationMessages(
func (h *Handler) GetWorkflowExecutionHistory(
ctx context.Context,
request *historyservice.GetWorkflowExecutionHistoryRequest,
) (_ *historyservice.GetWorkflowExecutionHistoryResponseWithRaw, retErr error) {
) (_ *historyservice.GetWorkflowExecutionHistoryResponse, retErr error) {
defer log.CapturePanic(h.logger, &retErr)
h.startWG.Wait()

2
service/history/history_engine.go
View File

@@ -1011,7 +1011,7 @@ func (e *historyEngineImpl) GetReplicationStatus(
func (e *historyEngineImpl) GetWorkflowExecutionHistory(
ctx context.Context,
request *historyservice.GetWorkflowExecutionHistoryRequest,
) (_ *historyservice.GetWorkflowExecutionHistoryResponseWithRaw, retError error) {
) (_ *historyservice.GetWorkflowExecutionHistoryResponse, retError error) {
return getworkflowexecutionhistory.Invoke(ctx, e.shardContext, e.workflowConsistencyChecker, e.versionChecker, e.eventNotifier, request, e.persistenceVisibilityMgr)
}

94
service/history/history_engine_test.go
View File

@@ -5478,11 +5478,6 @@ func (s *engineSuite) TestGetHistory() {
func (s *engineSuite) TestGetWorkflowExecutionHistory() {
we := commonpb.WorkflowExecution{WorkflowId: "wid1", RunId: uuid.New()}
namespaceEntry := namespace.NewLocalNamespaceForTest(
&persistencespb.NamespaceInfo{Name: "test-namespace"},
&persistencespb.NamespaceConfig{},
"")
s.mockNamespaceCache.EXPECT().GetNamespaceByID(gomock.Any()).Return(namespaceEntry, nil).AnyTimes()
newRunID := uuid.New()
req := &historyservice.GetWorkflowExecutionHistoryRequest{
@@ -5528,8 +5523,15 @@ func (s *engineSuite) TestGetWorkflowExecutionHistory() {
RunID: we.RunId,
}).Return(&persistence.GetWorkflowExecutionResponse{State: mState}, nil).AnyTimes()
// GetWorkflowExecutionHistory will request the last event
history := historypb.History{
Events: []*historypb.HistoryEvent{
s.mockExecutionMgr.EXPECT().ReadHistoryBranch(gomock.Any(), &persistence.ReadHistoryBranchRequest{
BranchToken: branchToken,
MinEventID: 5,
MaxEventID: 6,
PageSize: 10,
NextPageToken: nil,
ShardID: 1,
}).Return(&persistence.ReadHistoryBranchResponse{
HistoryEvents: []*historypb.HistoryEvent{
{
EventId: int64(5),
EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED,
@@ -5543,28 +5545,9 @@ func (s *engineSuite) TestGetWorkflowExecutionHistory() {
},
},
},
}
historyBlob, err := history.Marshal()
s.NoError(err)
s.mockExecutionMgr.EXPECT().ReadRawHistoryBranch(gomock.Any(), &persistence.ReadHistoryBranchRequest{
BranchToken: branchToken,
MinEventID: 5,
MaxEventID: 6,
PageSize: 10,
NextPageToken: nil,
ShardID: 1,
}).Return(&persistence.ReadRawHistoryBranchResponse{
HistoryEventBlobs: []*commonpb.DataBlob{
{
EncodingType: enumspb.ENCODING_TYPE_PROTO3,
Data: historyBlob,
},
},
NextPageToken: []byte{},
Size: 1,
}, nil).Times(1)
}, nil).Times(2)
s.mockExecutionMgr.EXPECT().TrimHistoryBranch(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
s.mockSearchAttributesProvider.EXPECT().GetSearchAttributes(gomock.Any(), false).Return(searchattribute.TestNameTypeMap, nil).AnyTimes()
@@ -5573,18 +5556,35 @@ func (s *engineSuite) TestGetWorkflowExecutionHistory() {
engine, err := s.historyEngine.shardContext.GetEngine(context.Background())
s.NoError(err)
resp, err := engine.GetWorkflowExecutionHistory(context.Background(), req)
oldGoSDKVersion := "1.9.1"
newGoSDKVersion := "1.10.1"
// new sdk: should see failed event
ctx := headers.SetVersionsForTests(context.Background(), newGoSDKVersion, headers.ClientNameGoSDK, headers.SupportedServerVersions, headers.AllFeatures)
resp, err := engine.GetWorkflowExecutionHistory(ctx, req)
s.NoError(err)
s.False(resp.Response.Archived)
err = history.Unmarshal(resp.Response.History[0])
s.NoError(err)
event := history.Events[0]
event := resp.Response.History.Events[0]
s.Equal(int64(5), event.EventId)
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED, event.EventType)
attrs := event.GetWorkflowExecutionFailedEventAttributes()
s.Equal("this workflow failed", attrs.Failure.Message)
s.Equal(newRunID, attrs.NewExecutionRunId)
s.Equal(enumspb.RETRY_STATE_IN_PROGRESS, attrs.RetryState)
// old sdk: should see continued-as-new event
// TODO: We can remove this once we no longer support SDK versions prior to around September 2021.
// See comment in workflowHandler.go:GetWorkflowExecutionHistory
ctx = headers.SetVersionsForTests(context.Background(), oldGoSDKVersion, headers.ClientNameGoSDK, headers.SupportedServerVersions, "")
resp, err = engine.GetWorkflowExecutionHistory(ctx, req)
s.NoError(err)
s.False(resp.Response.Archived)
event = resp.Response.History.Events[0]
s.Equal(int64(5), event.EventId)
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW, event.EventType)
attrs2 := event.GetWorkflowExecutionContinuedAsNewEventAttributes()
s.Equal(newRunID, attrs2.NewExecutionRunId)
s.Equal("this workflow failed", attrs2.Failure.Message)
}
func (s *engineSuite) TestGetWorkflowExecutionHistory_RawHistoryWithTransientDecision() {
@@ -5629,22 +5629,18 @@ func (s *engineSuite) TestGetWorkflowExecutionHistory_RawHistoryWithTransientDec
}
s.mockNamespaceCache.EXPECT().GetNamespaceID(tests.Namespace).Return(tests.NamespaceID, nil).AnyTimes()
historyBlob1, err := s.mockShard.GetPayloadSerializer().SerializeEvents(
[]*historypb.HistoryEvent{
{
EventId: int64(3),
EventType: enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED,
},
historyBlob1, err := s.mockShard.GetPayloadSerializer().SerializeEvent(
&historypb.HistoryEvent{
EventId: int64(3),
EventType: enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED,
},
enumspb.ENCODING_TYPE_PROTO3,
)
s.NoError(err)
historyBlob2, err := s.mockShard.GetPayloadSerializer().SerializeEvents(
[]*historypb.HistoryEvent{
{
EventId: int64(4),
EventType: enumspb.EVENT_TYPE_WORKFLOW_TASK_TIMED_OUT,
},
historyBlob2, err := s.mockShard.GetPayloadSerializer().SerializeEvent(
&historypb.HistoryEvent{
EventId: int64(4),
EventType: enumspb.EVENT_TYPE_WORKFLOW_TASK_TIMED_OUT,
},
enumspb.ENCODING_TYPE_PROTO3,
)
@@ -5666,12 +5662,14 @@ func (s *engineSuite) TestGetWorkflowExecutionHistory_RawHistoryWithTransientDec
resp, err := engine.GetWorkflowExecutionHistory(ctx, req)
s.NoError(err)
s.False(resp.Response.Archived)
s.Empty(resp.Response.History)
s.Len(resp.Response.RawHistory, 3)
historyEvents, err := s.mockShard.GetPayloadSerializer().DeserializeEvents(resp.Response.RawHistory[2])
s.Empty(resp.Response.History.Events)
s.Len(resp.Response.RawHistory, 4)
event, err := s.mockShard.GetPayloadSerializer().DeserializeEvent(resp.Response.RawHistory[2])
s.NoError(err)
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED, historyEvents[0].EventType)
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED, historyEvents[1].EventType)
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_TASK_SCHEDULED, event.EventType)
event, err = s.mockShard.GetPayloadSerializer().DeserializeEvent(resp.Response.RawHistory[3])
s.NoError(err)
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED, event.EventType)
}
func (s *engineSuite) Test_GetWorkflowExecutionRawHistoryV2_FailedOnInvalidWorkflowID() {

2
service/history/shard/engine.go
View File

@@ -109,7 +109,7 @@ type (
GetReplicationStatus(ctx context.Context, request *historyservice.GetReplicationStatusRequest) (*historyservice.ShardReplicationStatus, error)
UpdateWorkflowExecution(ctx context.Context, request *historyservice.UpdateWorkflowExecutionRequest) (*historyservice.UpdateWorkflowExecutionResponse, error)
PollWorkflowExecutionUpdate(ctx context.Context, request *historyservice.PollWorkflowExecutionUpdateRequest) (*historyservice.PollWorkflowExecutionUpdateResponse, error)
GetWorkflowExecutionHistory(ctx context.Context, request *historyservice.GetWorkflowExecutionHistoryRequest) (*historyservice.GetWorkflowExecutionHistoryResponseWithRaw, error)
GetWorkflowExecutionHistory(ctx context.Context, request *historyservice.GetWorkflowExecutionHistoryRequest) (*historyservice.GetWorkflowExecutionHistoryResponse, error)
GetWorkflowExecutionHistoryReverse(ctx context.Context, request *historyservice.GetWorkflowExecutionHistoryReverseRequest) (*historyservice.GetWorkflowExecutionHistoryReverseResponse, error)
GetWorkflowExecutionRawHistory(ctx context.Context, request *historyservice.GetWorkflowExecutionRawHistoryRequest) (*historyservice.GetWorkflowExecutionRawHistoryResponse, error)
GetWorkflowExecutionRawHistoryV2(ctx context.Context, request *historyservice.GetWorkflowExecutionRawHistoryV2Request) (*historyservice.GetWorkflowExecutionRawHistoryV2Response, error)

4
service/history/shard/engine_mock.go
View File

@@ -302,10 +302,10 @@ func (mr *MockEngineMockRecorder) GetReplicationTasksIter(ctx, pollingCluster, m
}
// GetWorkflowExecutionHistory mocks base method.
func (m *MockEngine) GetWorkflowExecutionHistory(ctx context.Context, request *historyservice.GetWorkflowExecutionHistoryRequest) (*historyservice.GetWorkflowExecutionHistoryResponseWithRaw, error) {
func (m *MockEngine) GetWorkflowExecutionHistory(ctx context.Context, request *historyservice.GetWorkflowExecutionHistoryRequest) (*historyservice.GetWorkflowExecutionHistoryResponse, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetWorkflowExecutionHistory", ctx, request)
ret0, _ := ret[0].(*historyservice.GetWorkflowExecutionHistoryResponseWithRaw)
ret0, _ := ret[0].(*historyservice.GetWorkflowExecutionHistoryResponse)
ret1, _ := ret[1].(error)
return ret0, ret1
}