NOT FOR REVIEW: PoC without MS Cache

common/cache/lru.go

@@ -63,6 +63,7 @@
 		pin            bool
 		timeSource     clock.TimeSource
 		metricsHandler metrics.Handler
+		enabled        bool
 	}
 
 	iteratorImpl struct {
@@ -78,6 +79,10 @@
 		refCount   int
 		size       int
 	}
+	//
+	//hasUpdate interface {
+	//	UpdateRegistry(ctx context.Context) update.Registry
+	//}
 )
 
 // Close closes the iterator
@@ -154,12 +159,12 @@
 
 // New creates a new cache with the given options
 func New(maxSize int, opts *Options) Cache {
-	return NewWithMetrics(maxSize, opts, metrics.NoopMetricsHandler)
+	return NewWithMetrics(maxSize, opts, metrics.NoopMetricsHandler, false)
 }
 
 // NewWithMetrics creates a new cache that will emit capacity and ttl metrics.
 // handler should be tagged with metrics.CacheTypeTag.
-func NewWithMetrics(maxSize int, opts *Options, handler metrics.Handler) Cache {
+func NewWithMetrics(maxSize int, opts *Options, handler metrics.Handler, enabled bool) Cache {
 	if opts == nil {
 		opts = &Options{}
 	}
@@ -181,6 +186,7 @@
 		onEvict:        opts.OnEvict,
 		timeSource:     timeSource,
 		metricsHandler: handler,
+		enabled:        enabled,
 	}
 }
 
@@ -273,6 +279,9 @@
 	if entry.refCount == 0 {
 		c.pinnedSize -= entry.Size()
 		metrics.CachePinnedUsage.With(c.metricsHandler).Record(float64(c.pinnedSize))
+		if !c.enabled {
+			c.tryEvictAndGetPreviousElement(entry, c.byKey[key])
+		}
 	}
 	// Entry size might have changed. Recalculate size and evict entries if necessary.
 	newEntrySize := getSize(entry.value)

common/cache/lru_test.go

@@ -57,7 +57,7 @@ func TestLRU(t *testing.T) {
 	metricsHandler := metricstest.NewCaptureHandler()
 	capture := metricsHandler.StartCapture()
 
-	cache := NewWithMetrics(4, nil, metricsHandler)
+	cache := NewWithMetrics(4, nil, metricsHandler, true)
 
 	cache.Put("A", "Foo")
 	assert.Equal(t, "Foo", cache.Get("A"))
@@ -148,6 +148,7 @@ func TestLRUWithTTL(t *testing.T) {
 			TimeSource: timeSource,
 		},
 		metricsHandler,
+		false,
 	)
 	cache.Put("A", "foo")
 	assert.Equal(t, "foo", cache.Get("A"))
@@ -249,6 +250,7 @@ func TestTTLWithPin(t *testing.T) {
 			TimeSource: timeSource,
 		},
 		metricsHandler,
+		false,
 	)
 
 	capture := metricsHandler.StartCapture()

common/dynamicconfig/constants.go

@@ -2626,4 +2626,9 @@ WorkerActivitiesPerSecond, MaxConcurrentActivityTaskPollers.
 		false,
 		`ActivityAPIsEnabled is a "feature enable" flag. `,
 	)
+	WorkflowCacheEnabled = NewGlobalBoolSetting(
+		"history.workflowCacheEnabled",
+		true,
+		``,
+	)
 )

common/nexus/endpoint_registry.go

@@ -126,7 +126,7 @@ func NewEndpointRegistry(
 		logger:          logger,
 		readThroughCacheByID: cache.NewWithMetrics(config.readThroughCacheSize(), &cache.Options{
 			TTL: config.readThroughCacheTTL(),
-		}, metricsHandler.WithTags(metrics.CacheTypeTag(metrics.NexusEndpointRegistryReadThroughCacheTypeTagValue))),
+		}, metricsHandler.WithTags(metrics.CacheTypeTag(metrics.NexusEndpointRegistryReadThroughCacheTypeTagValue)), true),
 	}
 }
 

service/history/configs/config.go

@@ -376,7 +376,8 @@ type Config struct {
 
 	BreakdownMetricsByTaskQueue dynamicconfig.BoolPropertyFnWithTaskQueueFilter
 
-	LogAllReqErrors dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	LogAllReqErrors      dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	WorkflowCacheEnabled dynamicconfig.BoolPropertyFn
 }
 
 // NewConfig returns new service config with default values
@@ -686,7 +687,8 @@ func NewConfig(
 
 		BreakdownMetricsByTaskQueue: dynamicconfig.MetricsBreakdownByTaskQueue.Get(dc),
 
-		LogAllReqErrors: dynamicconfig.LogAllReqErrors.Get(dc),
+		LogAllReqErrors:      dynamicconfig.LogAllReqErrors.Get(dc),
+		WorkflowCacheEnabled: dynamicconfig.WorkflowCacheEnabled.Get(dc),
 	}
 
 	return cfg

service/history/events/cache.go

@@ -110,7 +110,7 @@ func newEventsCache(
 
 	taggedMetricHandler := metricsHandler.WithTags(metrics.CacheTypeTag(metrics.EventsCacheTypeTagValue))
 	return &CacheImpl{
-		Cache:            cache.NewWithMetrics(maxSize, opts, taggedMetricHandler),
+		Cache:            cache.NewWithMetrics(maxSize, opts, taggedMetricHandler, true),
 		executionManager: executionManager,
 		metricsHandler:   taggedMetricHandler,
 		logger:           logger,

service/history/replication/progress_cache.go

@@ -82,7 +82,7 @@ func NewProgressCache(
 		TTL: config.ReplicationProgressCacheTTL(),
 	}
 	return &progressCacheImpl{
-		cache: cache.NewWithMetrics(maxSize, opts, handler.WithTags(metrics.CacheTypeTag(metrics.MutableStateCacheTypeTagValue))),
+		cache: cache.NewWithMetrics(maxSize, opts, handler.WithTags(metrics.CacheTypeTag(metrics.MutableStateCacheTypeTagValue)), true),
 	}
 }
 

service/history/workflow/cache/cache.go

@@ -125,6 +125,7 @@ func NewHostLevelCache(
 		config.HistoryCacheNonUserContextLockTimeout(),
 		logger,
 		handler,
+		config.WorkflowCacheEnabled(),
 	)
 }
 
@@ -143,6 +144,7 @@ func NewShardLevelCache(
 		config.HistoryCacheNonUserContextLockTimeout(),
 		logger,
 		handler,
+		config.WorkflowCacheEnabled(),
 	)
 }
 
@@ -152,6 +154,7 @@ func newCache(
 	nonUserContextLockTimeout time.Duration,
 	logger log.Logger,
 	handler metrics.Handler,
+	cacheEnabled bool,
 ) Cache {
 	opts := &cache.Options{
 		TTL: ttl,
@@ -193,7 +196,7 @@ func newCache(
 		},
 	}
 
-	withMetrics := cache.NewWithMetrics(size, opts, handler.WithTags(metrics.CacheTypeTag(metrics.MutableStateCacheTypeTagValue)))
+	withMetrics := cache.NewWithMetrics(size, opts, handler.WithTags(metrics.CacheTypeTag(metrics.MutableStateCacheTypeTagValue)), cacheEnabled)
 
 	return &cacheImpl{
 		Cache:                     withMetrics,