Byte size replication cache per shard (#7243)

<!-- Describe what has changed in this PR -->
**What changed?**
Enforced per-shard replication cache limits by item count and total byte
size.
Added ByteSize() implementations for ReplicationTask and all referenced
sub-types to enable accurate memory accounting.
Added util test methods to verify ByteSize implementation and ensure
that additions to structs using ByteSize will have ByteSize implemented
and implemented correctly.


<!-- Tell your future self why have you made these changes -->
**Why?**
Replication task sizes vary widely—from really small to hundreds of KiB.
A count-only limit lets a few large tasks blow past the intended memory
budget, leading to potential OOM in the history service. Adding a
byte-size bound (with accurate ByteSize() accounting) keeps each shard
within a predictable memory maximum usage.


<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
**How did you test it?**
Unit tests and deployment tests.


<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
**Potential risks**
Replication cache continues to be disabled by default. Enabling it
without a max size value or a value that is too large may cause memory
issues.

<!-- Is it notable for release? e.g. schema updates, configuration or
data migration required? If so, please mention it, and also update
CHANGELOG.md -->
**Release notes**
Introduce per-shard replication cache limits by count and total bytes
(opt-in). Implement ByteSize() across ReplicationTask and nested types
for accurate memory accounting, reducing risk of memory spikes in the
history service.

<!-- Is there any documentation updates should be made for config,
https://cadenceworkflow.io/docs/operation-guide/setup/ ? If so, please
open an PR in https://github.com/cadence-workflow/cadence-docs -->
**Documentation Changes**

---------

Signed-off-by: fimanishi <[email protected]>

Author: fimanishi

cmd/server/go.mod

@@ -79,6 +79,7 @@ require (
 	github.com/coreos/go-semver v0.3.0 // indirect
 	github.com/coreos/go-systemd/v22 v22.3.2 // indirect
 	github.com/fatih/color v1.13.0 // indirect
+	github.com/google/gofuzz v1.0.0 // indirect
 	github.com/mattn/go-colorable v0.1.9 // indirect
 	github.com/mattn/go-isatty v0.0.14 // indirect
 	github.com/ncruces/go-sqlite3 v0.22.0 // indirect

common/archiver/gcloud/go.mod

@@ -73,6 +73,7 @@ require (
 	github.com/golang-jwt/jwt/v5 v5.2.0 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/google/gofuzz v1.0.0 // indirect
 	github.com/google/s2a-go v0.1.4 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.2.4 // indirect
 	github.com/googleapis/gax-go/v2 v2.12.0 // indirect

common/dynamicconfig/dynamicproperties/constants.go

@@ -1091,9 +1091,15 @@ const (
 	// ReplicatorCacheCapacity is the capacity of replication cache in number of tasks
 	// KeyName: history.replicatorCacheCapacity
 	// Value type: Int
-	// Default value: 10000
+	// Default value: 0
 	// Allowed filters: N/A
 	ReplicatorCacheCapacity
+	// ReplicatorCacheMaxSize is the max size of the replication cache in bytes
+	// KeyName: history.replicatorCacheSize
+	// Value type: Int
+	// Default value: 0
+	// Allowed filters: N/A
+	ReplicatorCacheMaxSize
 
 	// ExecutionMgrNumConns is persistence connections number for ExecutionManager
 	// KeyName: history.executionMgrNumConns
@@ -3693,6 +3699,11 @@ var IntKeys = map[IntKey]DynamicInt{
 		Description:  "ReplicatorCacheCapacity is the capacity of replication cache in number of tasks",
 		DefaultValue: 0,
 	},
+	ReplicatorCacheMaxSize: {
+		KeyName:      "history.replicatorCacheSize",
+		Description:  "ReplicatorCacheMaxSize is the max size of the replication cache in bytes",
+		DefaultValue: 0,
+	},
 	ExecutionMgrNumConns: {
 		KeyName:      "history.executionMgrNumConns",
 		Description:  "Deprecated: not used. ExecutionMgrNumConns is persistence connections number for ExecutionManager",

common/types/admin.go

@@ -20,7 +20,10 @@
 
 package types
 
-import "sort"
+import (
+	"sort"
+	"unsafe"
+)
 
 // AddSearchAttributeRequest is an internal type (TBD...)
 type AddSearchAttributeRequest struct {
@@ -317,6 +320,14 @@ type IsolationGroupPartition struct {
 	State IsolationGroupState
 }
 
+// ByteSize returns an approximate size of the object in bytes
+func (i IsolationGroupPartition) ByteSize() uint64 {
+	var size uint64
+	size += uint64(unsafe.Sizeof(i))
+	size += uint64(len(i.Name))
+	return size
+}
+
 // IsolationGroupConfiguration is an internal representation of a set of
 // isolation-groups as a mapping and may refer to either globally or per-domain (or both) configurations.
 // and their statuses. It's redundantly indexed by IsolationGroup name to simplify lookups.
@@ -356,6 +367,19 @@ func (i IsolationGroupConfiguration) DeepCopy() IsolationGroupConfiguration {
 	return out
 }
 
+// ByteSize returns an approximate size of the object in bytes
+func (i *IsolationGroupConfiguration) ByteSize() uint64 {
+	if i == nil {
+		return 0
+	}
+
+	size := uint64(unsafe.Sizeof(*i))
+	for k, v := range *i {
+		size += uint64(len(k)) + uint64(len(v.Name))
+	}
+	return size
+}
+
 // FromIsolationGroupPartitionList maps a list of isolation to the internal IsolationGroup configuration type
 // whose map keys tend to be used more for set operations
 func FromIsolationGroupPartitionList(in []IsolationGroupPartition) IsolationGroupConfiguration {
@@ -422,6 +446,18 @@ func (c AsyncWorkflowConfiguration) DeepCopy() AsyncWorkflowConfiguration {
 	return res
 }
 
+// ByteSize returns an approximate size of the object in bytes
+func (c *AsyncWorkflowConfiguration) ByteSize() uint64 {
+	if c == nil {
+		return 0
+	}
+	size := uint64(unsafe.Sizeof(*c))
+	size += uint64(len(c.PredefinedQueueName))
+	size += uint64(len(c.QueueType))
+	size += c.QueueConfig.ByteSize()
+	return size
+}
+
 type UpdateDomainAsyncWorkflowConfiguratonRequest struct {
 	Domain        string
 	Configuration *AsyncWorkflowConfiguration

common/types/replicator.go

@@ -24,6 +24,7 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+	"unsafe"
 )
 
 // DLQType is an internal type (TBD...)
@@ -99,6 +100,14 @@ func (e DomainOperation) String() string {
 	return fmt.Sprintf("DomainOperation(%d)", w)
 }
 
+// ByteSize returns the approximate memory used in bytes
+func (e *DomainOperation) ByteSize() uint64 {
+	if e == nil {
+		return 0
+	}
+	return uint64(unsafe.Sizeof(*e))
+}
+
 // UnmarshalText parses enum value from string representation
 func (e *DomainOperation) UnmarshalText(value []byte) error {
 	switch s := strings.ToUpper(string(value)); s {
@@ -195,6 +204,22 @@ func (v *DomainTaskAttributes) GetPreviousFailoverVersion() (o int64) {
 	return
 }
 
+// ByteSize returns the approximate memory used in bytes
+func (v *DomainTaskAttributes) ByteSize() uint64 {
+	if v == nil {
+		return 0
+	}
+
+	size := uint64(unsafe.Sizeof(*v))
+	size += v.DomainOperation.ByteSize()
+	size += uint64(len(v.ID))
+	size += v.Info.ByteSize()
+	size += v.Config.ByteSize()
+	size += v.ReplicationConfig.ByteSize()
+
+	return size
+}
+
 // FailoverMarkerAttributes is an internal type (TBD...)
 type FailoverMarkerAttributes struct {
 	DomainID        string `json:"domainID,omitempty"`
@@ -226,6 +251,21 @@ func (v *FailoverMarkerAttributes) GetCreationTime() (o int64) {
 	return
 }
 
+// ByteSize returns the approximate memory used in bytes
+func (v *FailoverMarkerAttributes) ByteSize() uint64 {
+	if v == nil {
+		return 0
+	}
+
+	size := uint64(unsafe.Sizeof(*v))
+	size += uint64(len(v.DomainID))
+	if v.CreationTime != nil {
+		size += uint64(unsafe.Sizeof(*v.CreationTime))
+	}
+
+	return size
+}
+
 // FailoverMarkers is an internal type (TBD...)
 type FailoverMarkers struct {
 	FailoverMarkers []*FailoverMarkerAttributes `json:"failoverMarkers,omitempty"`
@@ -399,6 +439,28 @@ func (v *HistoryTaskV2Attributes) GetNewRunEvents() (o *DataBlob) {
 	return
 }
 
+// ByteSize returns the approximate memory used in bytes
+func (v *HistoryTaskV2Attributes) ByteSize() uint64 {
+	if v == nil {
+		return 0
+	}
+
+	size := uint64(unsafe.Sizeof(*v))
+	size += uint64(len(v.DomainID))
+	size += uint64(len(v.WorkflowID))
+	size += uint64(len(v.RunID))
+
+	size += uint64(len(v.VersionHistoryItems)) * uint64(unsafe.Sizeof((*VersionHistoryItem)(nil)))
+	for _, item := range v.VersionHistoryItems {
+		size += item.ByteSize()
+	}
+
+	size += v.Events.ByteSize()
+	size += v.NewRunEvents.ByteSize()
+
+	return size
+}
+
 type CountDLQMessagesRequest struct {
 	// ForceFetch will force fetching current values from DB
 	// instead of using cached values used for emitting metrics.
@@ -656,9 +718,12 @@ func (v *ReplicationMessages) GetEarliestCreationTime() *int64 {
 	return &result
 }
 
-// ReplicationMessagesSizeFn is a function type to calculate size of ReplicationMessages
+// ReplicationMessagesSizeFn is a function type to calculate the size of ReplicationMessages
 type ReplicationMessagesSizeFn func(v *ReplicationMessages) int
 
+// ReplicationTaskSizeFn is a function type to calculate the size of a single ReplicationTask
+type ReplicationTaskSizeFn func(v *ReplicationTask) int
+
 // ReplicationTask is an internal type (TBD...)
 type ReplicationTask struct {
 	TaskType                      *ReplicationTaskType           `json:"taskType,omitempty"`
@@ -727,6 +792,26 @@ func (v *ReplicationTask) GetCreationTime() (o int64) {
 	return
 }
 
+// ByteSize returns the approximate memory used in bytes
+func (v *ReplicationTask) ByteSize() uint64 {
+	if v == nil {
+		return 0
+	}
+
+	size := uint64(unsafe.Sizeof(*v))
+	size += v.TaskType.ByteSize()
+	size += v.DomainTaskAttributes.ByteSize()
+	size += v.SyncShardStatusTaskAttributes.ByteSize()
+	size += v.SyncActivityTaskAttributes.ByteSize()
+	size += v.HistoryTaskV2Attributes.ByteSize()
+	size += v.FailoverMarkerAttributes.ByteSize()
+	if v.CreationTime != nil {
+		size += uint64(unsafe.Sizeof(*v.CreationTime))
+	}
+
+	return size
+}
+
 // ReplicationTaskInfo is an internal type (TBD...)
 type ReplicationTaskInfo struct {
 	DomainID     string `json:"domainID,omitempty"`
@@ -898,6 +983,14 @@ const (
 	ReplicationTaskTypeFailoverMarker
 )
 
+// ByteSize returns the approximate memory used in bytes
+func (e *ReplicationTaskType) ByteSize() uint64 {
+	if e == nil {
+		return 0
+	}
+	return uint64(unsafe.Sizeof(*e))
+}
+
 // ReplicationToken is an internal type (TBD...)
 type ReplicationToken struct {
 	ShardID                int32 `json:"shardID,omitempty"`
@@ -1028,6 +1121,37 @@ func (v *SyncActivityTaskAttributes) GetVersionHistory() (o *VersionHistory) {
 	return
 }
 
+// ByteSize returns the approximate memory used in bytes
+func (v *SyncActivityTaskAttributes) ByteSize() uint64 {
+	if v == nil {
+		return 0
+	}
+
+	size := uint64(unsafe.Sizeof(*v))
+	size += uint64(len(v.DomainID))
+	size += uint64(len(v.WorkflowID))
+	size += uint64(len(v.RunID))
+	if v.ScheduledTime != nil {
+		size += uint64(unsafe.Sizeof(*v.ScheduledTime))
+	}
+	if v.StartedTime != nil {
+		size += uint64(unsafe.Sizeof(*v.StartedTime))
+	}
+	if v.LastHeartbeatTime != nil {
+		size += uint64(unsafe.Sizeof(*v.LastHeartbeatTime))
+	}
+	size += uint64(len(v.Details))
+	if v.LastFailureReason != nil {
+		size += uint64(unsafe.Sizeof(*v.LastFailureReason))
+		size += uint64(len(*v.LastFailureReason))
+	}
+	size += uint64(len(v.LastWorkerIdentity))
+	size += uint64(len(v.LastFailureDetails))
+	size += v.VersionHistory.ByteSize()
+
+	return size
+}
+
 // SyncShardStatus is an internal type (TBD...)
 type SyncShardStatus struct {
 	Timestamp *int64 `json:"timestamp,omitempty"`
@@ -1047,3 +1171,18 @@ type SyncShardStatusTaskAttributes struct {
 	ShardID       int64  `json:"shardId,omitempty"`
 	Timestamp     *int64 `json:"timestamp,omitempty"`
 }
+
+// ByteSize returns the approximate memory used in bytes
+func (v *SyncShardStatusTaskAttributes) ByteSize() uint64 {
+	if v == nil {
+		return 0
+	}
+
+	size := uint64(unsafe.Sizeof(*v))
+	size += uint64(len(v.SourceCluster))
+	if v.Timestamp != nil {
+		size += uint64(unsafe.Sizeof(*v.Timestamp))
+	}
+
+	return size
+}

common/types/replicator_test.go

@@ -1227,3 +1227,8 @@ func TestSyncShardStatus_GetTimestamp(t *testing.T) {
 	res = nilStruct.GetTimestamp()
 	assert.Equal(t, int64(0), res)
 }
+
+func TestReplicationTask_ByteSize(t *testing.T) {
+	AssertReachablesImplementByteSize(t, (*ReplicationTask)(nil))
+	AssertByteSizeMatchesReflect(t, &ReplicationTask{})
+}