Expose workflow history size and count to client (uber#5392)

<!-- Describe what has changed in this PR -->
**What changed?**
Added fields regarding workflow history size and count in PollForDecisionResponse to be exposed to cadence-client.
History size will be exposed in unit of bytes and history count will be the number of completed events.
Added fields sending history data from cadence-history to cadence-matching when matching service calls the persistence layer.
Updated dependency for cadence-idl to include new fields in idl in sync with cadence-client.

<!-- Tell your future self why have you made these changes -->
**Why?**
There have been requests from customers wanting history information regarding workflow to be exposed in client so that they can optimize their workflow through methods like continueAsNew when approaching the history size limit.
Resolved cadence-client issue [uber#917 ](uber-go/cadence-client#917)

<!-- How have you verified this change? Tested locally? Added a unit test? Checked in staging env? -->
**How did you test it?**
Standard unit test as well as tested in development environment.

<!-- Assuming the worst case, what can be broken when deploying this change to production? -->
**Potential risks**
Since we have added new fields to thrift and proto layer, there should not be any breaking change since new fields will be defaulted to 0. However, there remains the possibility that the change will result in thrift/proto field mismatch between serve and client hence result in information loss.

common/types/history.go

@@ -147,6 +147,7 @@ type GetMutableStateResponse struct {
 	WorkflowCloseState                   *int32             `json:"workflowCloseState,omitempty"`
 	VersionHistories                     *VersionHistories  `json:"versionHistories,omitempty"`
 	IsStickyTaskListEnabled              bool               `json:"isStickyTaskListEnabled,omitempty"`
+	HistorySize                          int64              `json:"historySize,omitempty"`
 }
 
 // GetNextEventID is an internal getter (TBD...)
@@ -585,6 +586,7 @@ type RecordDecisionTaskStartedResponse struct {
 	ScheduledTimestamp        *int64                    `json:"scheduledTimestamp,omitempty"`
 	StartedTimestamp          *int64                    `json:"startedTimestamp,omitempty"`
 	Queries                   map[string]*WorkflowQuery `json:"queries,omitempty"`
+	HistorySize               int64                     `json:"historySize,omitempty"`
 }
 
 // GetPreviousStartedEventID is an internal getter (TBD...)

common/types/mapper/proto/api.go

@@ -2191,6 +2191,7 @@ func FromPollForDecisionTaskResponse(t *types.PollForDecisionTaskResponse) *apiv
 		StartedTime:               unixNanoToTime(t.StartedTimestamp),
 		Queries:                   FromWorkflowQueryMap(t.Queries),
 		NextEventId:               t.NextEventID,
+		TotalHistoryBytes:         t.TotalHistoryBytes,
 	}
 }
 
@@ -2214,6 +2215,7 @@ func ToPollForDecisionTaskResponse(t *apiv1.PollForDecisionTaskResponse) *types.
 		StartedTimestamp:          timeToUnixNano(t.StartedTime),
 		Queries:                   ToWorkflowQueryMap(t.Queries),
 		NextEventID:               t.NextEventId,
+		TotalHistoryBytes:         t.TotalHistoryBytes,
 	}
 }
 

common/types/mapper/proto/history.go

@@ -337,6 +337,7 @@ func FromHistoryGetMutableStateResponse(t *types.GetMutableStateResponse) *histo
 		WorkflowCloseState:                   FromWorkflowExecutionCloseStatus(workflowCloseState),
 		VersionHistories:                     FromVersionHistories(t.VersionHistories),
 		IsStickyTaskListEnabled:              t.IsStickyTaskListEnabled,
+		HistorySize:                          t.HistorySize,
 	}
 }
 
@@ -363,6 +364,7 @@ func ToHistoryGetMutableStateResponse(t *historyv1.GetMutableStateResponse) *typ
 		VersionHistories:                     ToVersionHistories(t.VersionHistories),
 		IsStickyTaskListEnabled:              t.IsStickyTaskListEnabled,
 		IsWorkflowRunning:                    t.WorkflowState == sharedv1.WorkflowState_WORKFLOW_STATE_RUNNING,
+		HistorySize:                          t.HistorySize,
 	}
 }
 
@@ -893,6 +895,7 @@ func FromHistoryRecordDecisionTaskStartedResponse(t *types.RecordDecisionTaskSta
 		ScheduledTime:             unixNanoToTime(t.ScheduledTimestamp),
 		StartedTime:               unixNanoToTime(t.StartedTimestamp),
 		Queries:                   FromWorkflowQueryMap(t.Queries),
+		HistorySize:               t.HistorySize,
 	}
 }
 
@@ -915,6 +918,7 @@ func ToHistoryRecordDecisionTaskStartedResponse(t *historyv1.RecordDecisionTaskS
 		ScheduledTimestamp:        timeToUnixNano(t.ScheduledTime),
 		StartedTimestamp:          timeToUnixNano(t.StartedTime),
 		Queries:                   ToWorkflowQueryMap(t.Queries),
+		HistorySize:               t.HistorySize,
 	}
 }
 

common/types/mapper/proto/matching.go

@@ -420,6 +420,7 @@ func FromMatchingPollForDecisionTaskResponse(t *types.MatchingPollForDecisionTas
 		ScheduledTime:             unixNanoToTime(t.ScheduledTimestamp),
 		StartedTime:               unixNanoToTime(t.StartedTimestamp),
 		Queries:                   FromWorkflowQueryMap(t.Queries),
+		TotalHistoryBytes:         t.TotalHistoryBytes,
 	}
 }
 
@@ -445,6 +446,7 @@ func ToMatchingPollForDecisionTaskResponse(t *matchingv1.PollForDecisionTaskResp
 		ScheduledTimestamp:        timeToUnixNano(t.ScheduledTime),
 		StartedTimestamp:          timeToUnixNano(t.StartedTime),
 		Queries:                   ToWorkflowQueryMap(t.Queries),
+		TotalHistoryBytes:         t.TotalHistoryBytes,
 	}
 }
 

common/types/mapper/thrift/history.go

@@ -206,6 +206,7 @@ func FromGetMutableStateResponse(t *types.GetMutableStateResponse) *history.GetM
 		WorkflowCloseState:                   t.WorkflowCloseState,
 		VersionHistories:                     FromVersionHistories(t.VersionHistories),
 		IsStickyTaskListEnabled:              &t.IsStickyTaskListEnabled,
+		HistorySize:                          &t.HistorySize,
 	}
 }
 
@@ -233,6 +234,7 @@ func ToGetMutableStateResponse(t *history.GetMutableStateResponse) *types.GetMut
 		WorkflowCloseState:                   t.WorkflowCloseState,
 		VersionHistories:                     ToVersionHistories(t.VersionHistories),
 		IsStickyTaskListEnabled:              t.GetIsStickyTaskListEnabled(),
+		HistorySize:                          t.GetHistorySize(),
 	}
 }
 
@@ -628,6 +630,7 @@ func FromRecordDecisionTaskStartedResponse(t *types.RecordDecisionTaskStartedRes
 		ScheduledTimestamp:        t.ScheduledTimestamp,
 		StartedTimestamp:          t.StartedTimestamp,
 		Queries:                   FromWorkflowQueryMap(t.Queries),
+		HistorySize:               &t.HistorySize,
 	}
 }
 
@@ -651,6 +654,7 @@ func ToRecordDecisionTaskStartedResponse(t *history.RecordDecisionTaskStartedRes
 		ScheduledTimestamp:        t.ScheduledTimestamp,
 		StartedTimestamp:          t.StartedTimestamp,
 		Queries:                   ToWorkflowQueryMap(t.Queries),
+		HistorySize:               t.GetHistorySize(),
 	}
 }
 

common/types/mapper/thrift/matching.go

@@ -279,6 +279,7 @@ func FromMatchingPollForDecisionTaskResponse(t *types.MatchingPollForDecisionTas
 		ScheduledTimestamp:        t.ScheduledTimestamp,
 		StartedTimestamp:          t.StartedTimestamp,
 		Queries:                   FromWorkflowQueryMap(t.Queries),
+		TotalHistoryBytes:         &t.TotalHistoryBytes,
 	}
 }
 
@@ -305,6 +306,7 @@ func ToMatchingPollForDecisionTaskResponse(t *matching.PollForDecisionTaskRespon
 		ScheduledTimestamp:        t.ScheduledTimestamp,
 		StartedTimestamp:          t.StartedTimestamp,
 		Queries:                   ToWorkflowQueryMap(t.Queries),
+		TotalHistoryBytes:         t.GetTotalHistoryBytes(),
 	}
 }
 

common/types/mapper/thrift/shared.go

@@ -3590,6 +3590,7 @@ func FromPollForDecisionTaskResponse(t *types.PollForDecisionTaskResponse) *shar
 		StartedTimestamp:          t.StartedTimestamp,
 		Queries:                   FromWorkflowQueryMap(t.Queries),
 		NextEventId:               &t.NextEventID,
+		TotalHistoryBytes:         &t.TotalHistoryBytes,
 	}
 }
 
@@ -3614,6 +3615,7 @@ func ToPollForDecisionTaskResponse(t *shared.PollForDecisionTaskResponse) *types
 		StartedTimestamp:          t.StartedTimestamp,
 		Queries:                   ToWorkflowQueryMap(t.Queries),
 		NextEventID:               t.GetNextEventId(),
+		TotalHistoryBytes:         t.GetTotalHistoryBytes(),
 	}
 }
 

common/types/matching.go

@@ -413,6 +413,7 @@ type MatchingPollForDecisionTaskResponse struct {
 	ScheduledTimestamp        *int64                    `json:"scheduledTimestamp,omitempty"`
 	StartedTimestamp          *int64                    `json:"startedTimestamp,omitempty"`
 	Queries                   map[string]*WorkflowQuery `json:"queries,omitempty"`
+	TotalHistoryBytes         int64                     `json:"currentHistorySize,omitempty"`
 }
 
 // GetWorkflowExecution is an internal getter (TBD...)
@@ -455,6 +456,14 @@ func (v *MatchingPollForDecisionTaskResponse) GetBranchToken() (o []byte) {
 	return
 }
 
+// GetTotalHistoryBytes is an internal getter of returning the history size in bytes
+func (v *MatchingPollForDecisionTaskResponse) GetTotalHistoryBytes() (o int64) {
+	if v != nil {
+		return v.TotalHistoryBytes
+	}
+	return
+}
+
 // MatchingQueryWorkflowRequest is an internal type (TBD...)
 type MatchingQueryWorkflowRequest struct {
 	DomainUUID    string                `json:"domainUUID,omitempty"`

common/types/shared.go

@@ -4029,6 +4029,7 @@ type PollForDecisionTaskResponse struct {
 	StartedTimestamp          *int64                    `json:"startedTimestamp,omitempty"`
 	Queries                   map[string]*WorkflowQuery `json:"queries,omitempty"`
 	NextEventID               int64                     `json:"nextEventId,omitempty"`
+	TotalHistoryBytes         int64                     `json:"historySize,omitempty"`
 }
 
 // GetTaskToken is an internal getter (TBD...)
@@ -4079,6 +4080,13 @@ func (v *PollForDecisionTaskResponse) GetNextEventID() (o int64) {
 	return
 }
 
+func (v *PollForDecisionTaskResponse) GetHistorySize() (o int64) {
+	if v != nil {
+		return v.TotalHistoryBytes
+	}
+	return
+}
+
 // PollerInfo is an internal type (TBD...)
 type PollerInfo struct {
 	LastAccessTime *int64  `json:"lastAccessTime,omitempty"`
@@ -7163,7 +7171,7 @@ type WorkflowExecutionInfo struct {
 	StartTime         *int64                        `json:"startTime,omitempty"`
 	CloseTime         *int64                        `json:"closeTime,omitempty"`
 	CloseStatus       *WorkflowExecutionCloseStatus `json:"closeStatus,omitempty"`
-	HistoryLength     int64                         `json:"historyLength,omitempty"`
+	HistoryLength     int64                         `json:"historyLength,omitempty"` //should be history count
 	ParentDomainID    *string                       `json:"parentDomainId,omitempty"`
 	ParentDomain      *string                       `json:"parentDomain,omitempty"`
 	ParentExecution   *WorkflowExecution            `json:"parentExecution,omitempty"`

common/types/testdata/common.go

@@ -57,18 +57,19 @@ const (
 	ClientLibraryVersion = "ClientLibraryVersion"
 	SupportedVersions    = "SupportedVersions"
 
-	Attempt           = 2
-	PageSize          = 10
-	HistoryLength     = 20
-	BacklogCountHint  = 30
-	AckLevel          = 1001
-	ReadLevel         = 1002
-	RatePerSecond     = 3.14
-	TaskID            = 444
-	ShardID           = 12345
-	MessageID1        = 50001
-	MessageID2        = 50002
-	EventStoreVersion = 333
+	Attempt            = 2
+	PageSize           = 10
+	HistoryLength      = 20
+	BacklogCountHint   = 30
+	AckLevel           = 1001
+	ReadLevel          = 1002
+	RatePerSecond      = 3.14
+	TaskID             = 444
+	ShardID            = 12345
+	MessageID1         = 50001
+	MessageID2         = 50002
+	EventStoreVersion  = 333
+	HistorySizeInBytes = 77779
 
 	EventID1 = int64(1)
 	EventID2 = int64(2)

common/types/testdata/service_history.go

@@ -72,6 +72,7 @@ var (
 		WorkflowCloseState:                   common.Int32Ptr(persistence.WorkflowCloseStatusTimedOut),
 		VersionHistories:                     &VersionHistories,
 		IsStickyTaskListEnabled:              true,
+		HistorySize:                          HistorySizeInBytes,
 	}
 	HistoryGetReplicationMessagesRequest  = AdminGetReplicationMessagesRequest
 	HistoryGetReplicationMessagesResponse = AdminGetReplicationMessagesResponse
@@ -171,6 +172,7 @@ var (
 		ScheduledTimestamp:        &Timestamp1,
 		StartedTimestamp:          &Timestamp2,
 		Queries:                   WorkflowQueryMap,
+		HistorySize:               HistorySizeInBytes,
 	}
 	HistoryRefreshWorkflowTasksRequest = types.HistoryRefreshWorkflowTasksRequest{
 		DomainUIID: DomainID,

common/util.go

@@ -413,6 +413,7 @@ func CreateMatchingPollForDecisionTaskResponse(historyResponse *types.RecordDeci
 		ScheduledTimestamp:        historyResponse.ScheduledTimestamp,
 		StartedTimestamp:          historyResponse.StartedTimestamp,
 		Queries:                   historyResponse.Queries,
+		TotalHistoryBytes:         historyResponse.HistorySize,
 	}
 	if historyResponse.GetPreviousStartedEventID() != EmptyEventID {
 		matchingResp.PreviousStartedEventID = historyResponse.PreviousStartedEventID

go.mod

@@ -37,7 +37,7 @@ require (
 	github.com/sirupsen/logrus v1.8.1
 	github.com/stretchr/testify v1.8.1
 	github.com/uber-go/tally v3.3.15+incompatible
-	github.com/uber/cadence-idl v0.0.0-20230525234945-b6f203573446
+	github.com/uber/cadence-idl v0.0.0-20230905165949-03586319b849
 	github.com/uber/ringpop-go v0.8.5
 	github.com/uber/tchannel-go v1.22.2
 	github.com/urfave/cli v1.22.4

go.sum

@@ -558,8 +558,8 @@ github.com/uber-go/tally v3.3.12+incompatible/go.mod h1:YDTIBxdXyOU/sCWilKB4bgyu
 github.com/uber-go/tally v3.3.15+incompatible h1:9hLSgNBP28CjIaDmAuRTq9qV+UZY+9PcvAkXO4nNMwg=
 github.com/uber-go/tally v3.3.15+incompatible/go.mod h1:YDTIBxdXyOU/sCWilKB4bgyufu1cEi0jdVnRdxvjnmU=
 github.com/uber/cadence-idl v0.0.0-20211111101836-d6b70b60eb8c/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
-github.com/uber/cadence-idl v0.0.0-20230525234945-b6f203573446 h1:Z2qOF3Eu3SEKgn9u6HVNkWozcp0MDm3F0P8ZwFKzxvA=
-github.com/uber/cadence-idl v0.0.0-20230525234945-b6f203573446/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
+github.com/uber/cadence-idl v0.0.0-20230905165949-03586319b849 h1:j3bfADG1t35Lt4rNRpc9AuQ3l2cGw2Ao25Qt6rDamgc=
+github.com/uber/cadence-idl v0.0.0-20230905165949-03586319b849/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
 github.com/uber/jaeger-client-go v2.22.1+incompatible h1:NHcubEkVbahf9t3p75TOCR83gdUHXjRJvjoBh1yACsM=
 github.com/uber/jaeger-client-go v2.22.1+incompatible/go.mod h1:WVhlPFC8FDjOFMMWRy2pZqQJSXxYSwNYOkTr/Z6d3Kk=
 github.com/uber/jaeger-lib v2.2.0+incompatible h1:MxZXOiR2JuoANZ3J6DE/U0kSFv/eJ/GfSYVCjK7dyaw=

proto/internal/uber/cadence/history/v1/service.proto

@@ -349,6 +349,7 @@ message GetMutableStateResponse {
   api.v1.WorkflowExecutionCloseStatus workflow_close_state = 15;
   shared.v1.VersionHistories version_histories = 16;
   bool is_sticky_task_list_enabled = 17;
+  int64 history_size = 18;
 }
 
 message PollMutableStateRequest {
@@ -401,6 +402,7 @@ message RecordDecisionTaskStartedResponse {
   google.protobuf.Timestamp scheduled_time = 12;
   google.protobuf.Timestamp started_time = 13;
   map<string, api.v1.WorkflowQuery> queries = 14;
+  int64 history_size = 15;
 }
 
 message RecordActivityTaskStartedRequest {

proto/internal/uber/cadence/matching/v1/service.proto

@@ -111,6 +111,7 @@ message PollForDecisionTaskResponse {
   google.protobuf.Timestamp scheduled_time = 15;
   google.protobuf.Timestamp started_time = 16;
   map<string, api.v1.WorkflowQuery> queries = 17;
+  int64 total_history_bytes = 18;
 }
 
 message PollForActivityTaskRequest {

service/frontend/thriftHandler_test.go

@@ -143,7 +143,7 @@ func TestThriftHandler(t *testing.T) {
 	t.Run("PollForDecisionTask", func(t *testing.T) {
 		h.EXPECT().PollForDecisionTask(ctx, &types.PollForDecisionTaskRequest{}).Return(&types.PollForDecisionTaskResponse{}, internalErr).Times(1)
 		resp, err := th.PollForDecisionTask(ctx, &shared.PollForDecisionTaskRequest{})
-		assert.Equal(t, shared.PollForDecisionTaskResponse{StartedEventId: common.Int64Ptr(0), Attempt: common.Int64Ptr(0), BacklogCountHint: common.Int64Ptr(0), NextEventId: common.Int64Ptr(0)}, *resp)
+		assert.Equal(t, shared.PollForDecisionTaskResponse{StartedEventId: common.Int64Ptr(0), Attempt: common.Int64Ptr(0), BacklogCountHint: common.Int64Ptr(0), NextEventId: common.Int64Ptr(0), TotalHistoryBytes: common.Int64Ptr(0)}, *resp)
 		assert.Equal(t, expectedErr, err)
 	})
 	t.Run("QueryWorkflow", func(t *testing.T) {

service/frontend/workflowHandler.go

@@ -4354,6 +4354,7 @@ func (wh *WorkflowHandler) createPollForDecisionTaskResponse(
 		StartedTimestamp:          matchingResp.StartedTimestamp,
 		Queries:                   matchingResp.Queries,
 		NextEventID:               matchingResp.NextEventID,
+		TotalHistoryBytes:         matchingResp.TotalHistoryBytes,
 	}
 
 	return resp, nil

service/history/decision/handler.go

@@ -692,6 +692,7 @@ func (handler *handlerImpl) createRecordDecisionTaskStartedResponse(
 		queries[id] = input
 	}
 	response.Queries = queries
+	response.HistorySize = msBuilder.GetHistorySize()
 	return response, nil
 }
 

service/history/execution/context.go

@@ -237,6 +237,9 @@ func (c *contextImpl) GetHistorySize() int64 {
 
 func (c *contextImpl) SetHistorySize(size int64) {
 	c.stats.HistorySize = size
+	if c.mutableState != nil {
+		c.mutableState.SetHistorySize(size)
+	}
 }
 
 func (c *contextImpl) LoadExecutionStats(
@@ -313,6 +316,7 @@ func (c *contextImpl) LoadWorkflowExecutionWithTaskVersion(
 			Message: "workflowExecutionContext counter flushBeforeReady status after loading mutable state from DB",
 		}
 	}
+
 	return c.mutableState, nil
 }
 

service/history/execution/mutable_state.go

@@ -236,5 +236,8 @@ type (
 		StartTransaction(entry *cache.DomainCacheEntry, incomingTaskVersion int64) (bool, error)
 		CloseTransactionAsMutation(now time.Time, transactionPolicy TransactionPolicy) (*persistence.WorkflowMutation, []*persistence.WorkflowEvents, error)
 		CloseTransactionAsSnapshot(now time.Time, transactionPolicy TransactionPolicy) (*persistence.WorkflowSnapshot, []*persistence.WorkflowEvents, error)
+
+		GetHistorySize() int64
+		SetHistorySize(size int64)
 	}
 )

service/history/execution/mutable_state_builder.go

@@ -160,6 +160,8 @@ type (
 		logger                     log.Logger
 		metricsClient              metrics.Client
 		pendingActivityWarningSent bool
+
+		executionStats *persistence.ExecutionStats
 	}
 )
 
@@ -241,6 +243,7 @@ func newMutableStateBuilder(
 	s.taskGenerator = NewMutableStateTaskGenerator(shard.GetClusterMetadata(), shard.GetDomainCache(), s)
 	s.decisionTaskManager = newMutableStateDecisionTaskManager(s)
 
+	s.executionStats = &persistence.ExecutionStats{}
 	return s
 }
 
@@ -333,6 +336,7 @@ func (e *mutableStateBuilder) Load(
 	// TODO: remove this after all 2DC workflows complete
 	e.replicationState = state.ReplicationState
 	e.checksum = state.Checksum
+	e.executionStats = state.ExecutionStats
 
 	e.fillForBackwardsCompatibility()
 
@@ -4194,6 +4198,14 @@ func (e *mutableStateBuilder) UpdateDuplicatedResource(
 	e.appliedEvents[id] = struct{}{}
 }
 
+func (e *mutableStateBuilder) GetHistorySize() int64 {
+	return e.executionStats.HistorySize
+}
+
+func (e *mutableStateBuilder) SetHistorySize(size int64) {
+	e.executionStats.HistorySize = size
+}
+
 func (e *mutableStateBuilder) prepareCloseTransaction(
 	transactionPolicy TransactionPolicy,
 ) error {