support background entry removal for workflow cache (#7902)

## What changed?
This adds support to the lru cache to actively expire entries older than
the TTL, by spawning a background goroutine that periodically deletes
old entries. A dynamic config, off by default, is added that can enable
this feature for the workflow cache.

## Why?
This can reduce the memory usage, and associated Go GC resource usage,
for workflow entries that won't be utilized since they are past their
TTL.

## How did you test it?
- [x] built
- [x] run locally and tested manually
- [x] covered by existing tests
- [x] added new unit test(s)
- [ ] added new functional test(s)

Also running this in test setups to verify expected memory reduction.

## Potential risks

Author: alfred-landrum

common/cache/cache.go

@@ -4,6 +4,7 @@ import (
 	"time"
 
 	"go.temporal.io/server/common/clock"
+	"go.temporal.io/server/common/dynamicconfig"
 )
 
 // A Cache is a generalized interface to a cache.  See cache.LRU for a specific
@@ -34,6 +35,13 @@ type Cache interface {
 	Size() int
 }
 
+type StoppableCache interface {
+	Cache
+
+	// Stop halts any background processing, and should be called when the cache will no longer be used.
+	Stop()
+}
+
 // Options control the behavior of the cache.
 type Options struct {
 	// TTL controls the time-to-live for a given cache entry.  Cache entries that
@@ -49,6 +57,9 @@ type Options struct {
 	OnPut func(val any)
 
 	OnEvict func(val any)
+
+	// BackgroundEvict configures background scanning for expired entries.
+	BackgroundEvict func() dynamicconfig.CacheBackgroundEvictSettings
 }
 
 // SimpleOptions provides options that can be used to configure SimpleCache.

common/cache/lru.go

@@ -2,12 +2,15 @@ package cache
 
 import (
 	"container/list"
+	"context"
 	"sync"
 	"time"
 
 	enumspb "go.temporal.io/api/enums/v1"
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/server/common/clock"
+	"go.temporal.io/server/common/dynamicconfig"
+	"go.temporal.io/server/common/goro"
 	"go.temporal.io/server/common/metrics"
 )
 
@@ -27,18 +30,20 @@ const emptyEntrySize = 0
 // lru is a concurrent fixed size cache that evicts elements in lru order
 type (
 	lru struct {
-		mut            sync.Mutex
-		byAccess       *list.List
-		byKey          map[interface{}]*list.Element
-		maxSize        int
-		currSize       int
-		pinnedSize     int
-		onPut          func(val any)
-		onEvict        func(val any)
-		ttl            time.Duration
-		pin            bool
-		timeSource     clock.TimeSource
-		metricsHandler metrics.Handler
+		mut             sync.Mutex
+		byAccess        *list.List
+		byKey           map[interface{}]*list.Element
+		maxSize         int
+		currSize        int
+		pinnedSize      int
+		onPut           func(val any)
+		onEvict         func(val any)
+		ttl             time.Duration
+		pin             bool
+		timeSource      clock.TimeSource
+		metricsHandler  metrics.Handler
+		backgroundEvict dynamicconfig.TypedPropertyFn[dynamicconfig.CacheBackgroundEvictSettings]
+		loops           goro.Group
 	}
 
 	iteratorImpl struct {
@@ -129,40 +134,53 @@ func (entry *entryImpl) CreateTime() time.Time {
 }
 
 // New creates a new cache with the given options
-func New(maxSize int, opts *Options) Cache {
+func New(maxSize int, opts *Options) StoppableCache {
 	return NewWithMetrics(maxSize, opts, metrics.NoopMetricsHandler)
 }
 
 // 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) StoppableCache {
 	if opts == nil {
 		opts = &Options{}
 	}
+	if opts.BackgroundEvict == nil {
+		opts.BackgroundEvict = func() dynamicconfig.CacheBackgroundEvictSettings {
+			return dynamicconfig.CacheBackgroundEvictSettings{
+				Enabled: false,
+			}
+		}
+	}
+
 	timeSource := opts.TimeSource
 	if timeSource == nil {
 		timeSource = clock.NewRealTimeSource()
 	}
 
 	metrics.CacheSize.With(handler).Record(float64(maxSize))
 	metrics.CacheTtl.With(handler).Record(opts.TTL)
-	return &lru{
-		byAccess:       list.New(),
-		byKey:          make(map[interface{}]*list.Element),
-		ttl:            opts.TTL,
-		maxSize:        maxSize,
-		currSize:       0,
-		pin:            opts.Pin,
-		onPut:          opts.OnPut,
-		onEvict:        opts.OnEvict,
-		timeSource:     timeSource,
-		metricsHandler: handler,
-	}
+	c := &lru{
+		byAccess:        list.New(),
+		byKey:           make(map[interface{}]*list.Element),
+		ttl:             opts.TTL,
+		maxSize:         maxSize,
+		currSize:        0,
+		pin:             opts.Pin,
+		onPut:           opts.OnPut,
+		onEvict:         opts.OnEvict,
+		timeSource:      timeSource,
+		metricsHandler:  handler,
+		backgroundEvict: opts.BackgroundEvict,
+	}
+	if c.backgroundEvict().Enabled {
+		c.loops.Go(c.bgEvictLoop)
+	}
+	return c
 }
 
 // NewLRU creates a new LRU cache of the given size, setting initial capacity
 // to the max size
-func NewLRU(maxSize int, handler metrics.Handler) Cache {
+func NewLRU(maxSize int, handler metrics.Handler) StoppableCache {
 	return New(maxSize, nil)
 }
 
@@ -181,14 +199,14 @@ func (c *lru) Get(key interface{}) interface{} {
 
 	entry := element.Value.(*entryImpl)
 
-	metrics.CacheEntryAgeOnGet.With(c.metricsHandler).Record(c.timeSource.Now().UTC().Sub(entry.createTime))
-
 	if c.isEntryExpired(entry, c.timeSource.Now().UTC()) {
 		// Entry has expired
 		c.deleteInternal(element)
 		return nil
 	}
 
+	metrics.CacheEntryAgeOnGet.With(c.metricsHandler).Record(c.timeSource.Now().UTC().Sub(entry.createTime))
+
 	c.updateEntryRefCount(entry)
 	c.byAccess.MoveToFront(element)
 	return entry.value
@@ -424,3 +442,54 @@ func (c *lru) updateEntryRefCount(entry *entryImpl) {
 		}
 	}
 }
+
+func (c *lru) Stop() {
+	c.loops.Cancel()
+}
+
+func (c *lru) bgEvictLoop(ctx context.Context) error {
+	ch, t := c.timeSource.NewTimer(c.backgroundEvict().LoopInterval)
+	for {
+		select {
+		case <-ch:
+			settings := c.backgroundEvict()
+			if settings.Enabled {
+				c.bgEvict(settings)
+			}
+			t.Reset(settings.LoopInterval)
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}
+
+func (c *lru) bgEvict(settings dynamicconfig.CacheBackgroundEvictSettings) {
+	now := c.timeSource.Now().UTC()
+
+	// Limit each iteration to scanning MaxEntryPerCall entries, to avoid holding the cache lock for too long.
+	evictToMax := func() (again bool) {
+		c.mut.Lock()
+		defer c.mut.Unlock()
+
+		element := c.byAccess.Back()
+		if settings.MaxEntryPerCall <= 0 {
+			return false
+		}
+		for n := 0; n < settings.MaxEntryPerCall; n++ {
+			if element == nil {
+				return false
+			}
+			elementPrev := element.Prev()
+			entry := element.Value.(*entryImpl) // nolint:revive
+			if !c.isEntryExpired(entry, now) {
+				return false
+			}
+			c.deleteInternal(element)
+			element = elementPrev
+		}
+		return element != nil
+	}
+
+	for evictToMax() {
+	}
+}

common/cache/lru_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.temporal.io/server/common/clock"
+	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/metrics/metricstest"
 )
@@ -138,8 +139,7 @@ func TestLRUWithTTL(t *testing.T) {
 	assert.Equal(t, 2, len(snapshot[metrics.CacheUsage.Name()]))
 	assert.Equal(t, float64(0), snapshot[metrics.CacheUsage.Name()][1].Value)
 	assert.Equal(t, 0, cache.Size())
-	assert.Equal(t, 2, len(snapshot[metrics.CacheEntryAgeOnGet.Name()]))
-	assert.Equal(t, time.Millisecond*300, snapshot[metrics.CacheEntryAgeOnGet.Name()][1].Value)
+	assert.Equal(t, 1, len(snapshot[metrics.CacheEntryAgeOnGet.Name()]))
 	assert.Equal(t, time.Millisecond*300, snapshot[metrics.CacheEntryAgeOnEviction.Name()][0].Value)
 }
 
@@ -735,3 +735,117 @@ func TestCache_InvokeLifecycleCallbacks(t *testing.T) {
 	assert.Nil(t, cache.Get("key"))
 	require.Equal(t, 2, onEvict, "expected OnEvict callback to be invoked")
 }
+
+func TestCache_UnusedExpiry(t *testing.T) {
+	t.Parallel()
+	r := require.New(t)
+
+	ttl := 10 * time.Minute
+	loopInterval := 1 * time.Minute
+	timeSource := clock.NewEventTimeSource()
+
+	cache := New(5,
+		&Options{
+			TTL:        ttl,
+			TimeSource: timeSource,
+			BackgroundEvict: func() dynamicconfig.CacheBackgroundEvictSettings {
+				return dynamicconfig.CacheBackgroundEvictSettings{
+					Enabled:         true,
+					LoopInterval:    loopInterval,
+					MaxEntryPerCall: 1,
+				}
+			},
+		},
+	)
+
+	cache.Put(1, 1)
+	r.Equal(1, cache.Size())
+
+	r.Eventually(func() bool {
+		timeSource.Advance(loopInterval)
+		return cache.Size() == 0
+	}, 2*time.Second, 100*time.Millisecond)
+
+	cache.Put(2, 2)
+	timeSource.Advance(ttl / 2)
+	cache.Put(3, 3)
+	r.Equal(2, cache.Size())
+
+	r.Eventually(func() bool {
+		timeSource.Advance(loopInterval)
+		return cache.Size() == 1 && cache.Get(2) == nil && cache.Get(3) == 3
+	}, 2*time.Second, 100*time.Millisecond)
+
+	r.Eventually(func() bool {
+		timeSource.Advance(loopInterval)
+		return cache.Size() == 0 && cache.Get(2) == nil && cache.Get(3) == nil
+	}, 2*time.Second, 100*time.Millisecond)
+
+	// Stop the background goroutine, confirm no active expiration.
+	cache.Put(4, 4)
+	cache.Stop()
+	l, ok := cache.(*lru)
+	r.True(ok)
+	c := make(chan struct{})
+	go func() {
+		l.loops.Wait()
+		close(c)
+	}()
+	r.Eventually(func() bool {
+		select {
+		case <-c:
+			return true
+		default:
+			return false
+		}
+	}, 2*time.Second, 100*time.Millisecond)
+	timeSource.Advance(ttl + 1*time.Second)
+	// The cache should still have entry 4,
+	r.Equal(1, cache.Size())
+	// but this Get call will check the (hard) ttl & expire it.
+	r.Equal(nil, cache.Get(4))
+}
+
+func TestCache_UnusedExpiryPin(t *testing.T) {
+	t.Parallel()
+	r := require.New(t)
+
+	ttl := 10 * time.Minute
+	loopInterval := 1 * time.Minute
+	timeSource := clock.NewEventTimeSource()
+
+	cache := New(5,
+		&Options{
+			TTL:        ttl,
+			Pin:        true,
+			TimeSource: timeSource,
+			BackgroundEvict: func() dynamicconfig.CacheBackgroundEvictSettings {
+				return dynamicconfig.CacheBackgroundEvictSettings{
+					Enabled:         true,
+					LoopInterval:    loopInterval,
+					MaxEntryPerCall: 1,
+				}
+			},
+		},
+	)
+
+	_, err := cache.PutIfNotExist(1, 1)
+	r.NoError(err)
+	timeSource.Advance(ttl / 2)
+	cache.Release(1)
+	_, err = cache.PutIfNotExist(2, 2)
+	r.NoError(err)
+	r.Equal(2, cache.Size())
+
+	r.Eventually(func() bool {
+		timeSource.Advance(loopInterval)
+		return cache.Size() == 1 && cache.Get(1) == nil
+	}, 1*time.Second, 100*time.Millisecond)
+
+	cache.Release(2)
+
+	r.Eventually(func() bool {
+		timeSource.Advance(loopInterval)
+		return cache.Size() == 0
+	}, 1*time.Second, 100*time.Millisecond)
+}

common/dynamicconfig/constants.go

@@ -1400,6 +1400,11 @@ will wait on workflow lock acquisition. Requires service restart to take effect.
 		`HistoryCacheHostLevelMaxSizeBytes is the maximum size of the host level history cache. This is only used if
 HistoryCacheSizeBasedLimit is set to true.`,
 	)
+	HistoryCacheBackgroundEvict = NewGlobalTypedSetting(
+		"history.cacheBackgroundEvict",
+		DefaultHistoryCacheBackgroundEvictSettings,
+		`HistoryCacheBackgroundEvict configures background processing to purge expired entries from the history cache.`,
+	)
 	EnableWorkflowExecutionTimeoutTimer = NewGlobalBoolSetting(
 		"history.enableWorkflowExecutionTimeoutTimer",
 		true,

common/dynamicconfig/shared_constants.go

@@ -100,3 +100,20 @@ type CircuitBreakerSettings struct {
 	// Timeout: Period of open state before changing to half-open state (default 60s).`
 	Timeout time.Duration
 }
+
+type CacheBackgroundEvictSettings struct {
+	// Enabled controls whether background purging of expired entries is active. To enable,
+	// this must be set to true at process start, but can be dynamically set to false to
+	// stop scanning entries.
+	Enabled bool
+	// LoopInterval is the frequency that a background goroutine scans for expired entries.
+	LoopInterval time.Duration
+	// MaxEntryPerCall is the max number of entries that are scanned while the cache is locked.
+	MaxEntryPerCall int
+}
+
+var DefaultHistoryCacheBackgroundEvictSettings = CacheBackgroundEvictSettings{
+	Enabled:         false,
+	LoopInterval:    1 * time.Minute,
+	MaxEntryPerCall: 1024,
+}