Refactor XDC cache (temporalio#6583)

xwduan · web-flow · commit fa88fbfdc406 · 2024-10-03T10:39:28.000-07:00
## What changed?
&lt;!-- Describe what has changed in this PR --&gt;
1. Remove MaxEventId from cache key
2. Add NextEventID to cache value
## Why?
&lt;!-- Tell your future self why have you made these changes --&gt;
For State based replication usage. So the events can be retrieved from
cache
## How did you test it?
&lt;!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? --&gt;
current unit test
## Potential risks
&lt;!-- Assuming the worst case, what can be broken when deploying this
change to production? --&gt;
cache may stop working.
## Documentation
&lt;!-- Have you made sure this change doesn't falsify anything currently
stated in `docs/`? If significant
new behavior is added, have you described that in `docs/`? --&gt;
n/a
## Is hotfix candidate?
&lt;!-- Is this PR a hotfix candidate or does it require a notification to
be sent to the broader community? (Yes/No) --&gt;
no
diff --git a/common/persistence/client/fx.go b/common/persistence/client/fx.go
@@ -100,10 +100,12 @@ func ClusterNameProvider(config *cluster.Config) ClusterName {
 
 func EventBlobCacheProvider(
 	dc *dynamicconfig.Collection,
+	logger log.Logger,
 ) persistence.XDCCache {
 	return persistence.NewEventsBlobCache(
 		dynamicconfig.XDCCacheMaxSizeBytes.Get(dc)(),
 		20*time.Second,
+		logger,
 	)
 }
 
diff --git a/common/persistence/execution_manager.go b/common/persistence/execution_manager.go
@@ -477,12 +477,12 @@ func (m *executionManagerImpl) serializeWorkflowEventBatches(
 		xdcKVs[NewXDCCacheKey(
 			definition.NewWorkflowKey(workflowEvents.NamespaceID, workflowEvents.WorkflowID, workflowEvents.RunID),
 			workflowEvents.Events[0].EventId,
-			workflowEvents.Events[len(workflowEvents.Events)-1].EventId+1,
 			workflowEvents.Events[0].Version,
 		)] = NewXDCCacheValue(
 			baseWorkflowInfo,
 			versionHistoryItems,
 			[]*commonpb.DataBlob{newEvents.Node.Events},
+			workflowEvents.Events[len(workflowEvents.Events)-1].EventId+1,
 		)
 		newEvents.ShardID = shardID
 		workflowNewEvents = append(workflowNewEvents, newEvents)
diff --git a/common/persistence/xdc_cache.go b/common/persistence/xdc_cache.go
@@ -25,28 +25,33 @@
 package persistence
 
 import (
+	"fmt"
 	"time"
 
 	commonpb "go.temporal.io/api/common/v1"
+	historypb "go.temporal.io/api/history/v1"
 	historyspb "go.temporal.io/server/api/history/v1"
 	persistencepb "go.temporal.io/server/api/persistence/v1"
 	workflowspb "go.temporal.io/server/api/workflow/v1"
 	"go.temporal.io/server/common/cache"
 	"go.temporal.io/server/common/definition"
+	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/log/tag"
+	"go.temporal.io/server/common/persistence/serialization"
 	"go.temporal.io/server/common/persistence/versionhistory"
 )
 
 type (
 	XDCCacheKey struct {
 		WorkflowKey definition.WorkflowKey
 		MinEventID  int64 // inclusive
-		MaxEventID  int64 // exclusive
 		Version     int64
 	}
 	XDCCacheValue struct {
 		BaseWorkflowInfo    *workflowspb.BaseExecutionInfo
 		VersionHistoryItems []*historyspb.VersionHistoryItem
 		EventBlobs          []*commonpb.DataBlob
+		NextEventID         int64
 	}
 
 	XDCCache interface {
@@ -55,7 +60,9 @@ type (
 	}
 
 	XDCCacheImpl struct {
-		cache cache.Cache
+		cache      cache.Cache
+		logger     log.Logger
+		serializer serialization.Serializer
 	}
 )
 
@@ -69,13 +76,11 @@ var _ cache.SizeGetter = XDCCacheValue{}
 func NewXDCCacheKey(
 	workflowKey definition.WorkflowKey,
 	minEventID int64,
-	maxEventID int64,
 	version int64,
 ) XDCCacheKey {
 	return XDCCacheKey{
 		WorkflowKey: workflowKey,
 		MinEventID:  minEventID,
-		MaxEventID:  maxEventID,
 		Version:     version,
 	}
 }
@@ -84,11 +89,13 @@ func NewXDCCacheValue(
 	baseWorkflowInfo *workflowspb.BaseExecutionInfo,
 	versionHistoryItems []*historyspb.VersionHistoryItem,
 	eventBlobs []*commonpb.DataBlob,
+	nextEventID int64,
 ) XDCCacheValue {
 	return XDCCacheValue{
 		BaseWorkflowInfo:    baseWorkflowInfo,
 		VersionHistoryItems: versionHistoryItems,
 		EventBlobs:          eventBlobs,
+		NextEventID:         nextEventID,
 	}
 }
 
@@ -106,6 +113,7 @@ func (v XDCCacheValue) CacheSize() int {
 func NewEventsBlobCache(
 	maxBytes int,
 	ttl time.Duration,
+	logger log.Logger,
 ) *XDCCacheImpl {
 	return &XDCCacheImpl{
 		cache: cache.New(
@@ -115,13 +123,31 @@ func NewEventsBlobCache(
 				Pin: false,
 			},
 		),
+		logger:     logger,
+		serializer: serialization.NewSerializer(),
 	}
 }
 
 func (e *XDCCacheImpl) Put(
 	key XDCCacheKey,
 	value XDCCacheValue,
 ) {
+	existingValue, found := e.Get(key)
+	if found && existingValue.NextEventID != value.NextEventID {
+		deserializeBlobs := func(blobs []*commonpb.DataBlob) [][]*historypb.HistoryEvent {
+			events := make([][]*historypb.HistoryEvent, len(blobs))
+			for i, blob := range blobs {
+				var err error
+				events[i], err = e.serializer.DeserializeEvents(blob)
+				if err != nil {
+					e.logger.Error("Error deserializing events", tag.Error(err))
+					return nil
+				}
+			}
+			return events
+		}
+		e.logger.Error(fmt.Sprintf("Putting duplicate key in XDC cache: wf-key: %v, existing event blobs: %v, new event blobs: %v", key.WorkflowKey, deserializeBlobs(existingValue.EventBlobs), deserializeBlobs(value.EventBlobs)))
+	}
 	e.cache.Put(key, value)
 }
 
diff --git a/service/history/replication/raw_task_converter.go b/service/history/replication/raw_task_converter.go
@@ -420,7 +420,6 @@ func getVersionHistoryAndEvents(
 		if xdcCacheValue, ok := eventBlobCache.Get(persistence.NewXDCCacheKey(
 			workflowKey,
 			firstEventID,
-			nextEventID,
 			eventVersion,
 		)); ok {
 			return xdcCacheValue.VersionHistoryItems, xdcCacheValue.EventBlobs, xdcCacheValue.BaseWorkflowInfo, nil

Original file line number	Diff line number	Diff line change
`@@ -100,10 +100,12 @@ func ClusterNameProvider(config *cluster.Config) ClusterName {`
`100`	`100`
`101`	`101`	`func EventBlobCacheProvider(`
`102`	`102`	`dc *dynamicconfig.Collection,`
	`103`	`+ logger log.Logger,`
`103`	`104`	`) persistence.XDCCache {`
`104`	`105`	`return persistence.NewEventsBlobCache(`
`105`	`106`	`dynamicconfig.XDCCacheMaxSizeBytes.Get(dc)(),`
`106`	`107`	`20*time.Second,`
	`108`	`+ logger,`
`107`	`109`	`)`
`108`	`110`	`}`
`109`	`111`