VersionMembershipCache: Metrics and refactorings! (temporalio#8894)

## What changed?
- Added metrics, such as cache hits and cache misses, so that we can
understand if the currently set TTL for this cache (of 1 second) is too
low or too high.
- Also did some re-factorings: While working on this, I realized that it
was much simpler to add a new wrapper with a metrics handler, that
specifically served the use case of understanding cache hits and missed,
rather than use any of the existing implementations of caches that have
a metrics handler attached to them. (See *NewWithMetrics*)
- Thus, I took some inspiration from the way *newEventsCache* was
implemented and came up with this.

## Why?
- Explained above.

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

## Potential risks
- None.


<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Introduces a typed `VersionMembershipCache` wrapper with metrics and
refactors callers to use it.
> 
> - New `common/worker_versioning/version_membership_cache.go`: typed
cache API with `Get`/`Put`, emits `cache_requests` and `cache_miss` with
`cache_type=version_membership` and op scopes
`VersionMembershipCacheGet`/`Put`
> - Metrics: add `VersionMembershipCacheTypeTagValue` and new operation
scopes in `metrics/metric_defs.go`
> - Refactor: replace `cache.Cache` with `VersionMembershipCache` in
worker versioning validation and history APIs (`startworkflow`,
`signalwithstartworkflow`, `resetworkflow`, `multioperation`,
`updateworkflowoptions`), update DI to provide wrapped cache in
`service/history/fx.go`
> - History engine constructors and factory updated to accept the new
interface
> - Tests: add lightweight in-memory and noop implementations; update
unit tests to use the new interface
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
1ce3a10. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: Shivs11

common/metrics/metric_defs.go

@@ -43,6 +43,7 @@ const (
 
 	MutableStateCacheTypeTagValue                     = "mutablestate"
 	EventsCacheTypeTagValue                           = "events"
+	VersionMembershipCacheTypeTagValue                = "version_membership"
 	NexusEndpointRegistryReadThroughCacheTypeTagValue = "nexus_endpoint_registry_readthrough"
 
 	InvalidHistoryURITagValue    = "invalid_history_uri"
@@ -450,6 +451,10 @@ const (
 	OperationMemoryScheduledQueueProcessorScope = "MemoryScheduledQueueProcessor"
 	// OperationOutboundQueueProcessorScope is a scope for the outbound queue processor.
 	OperationOutboundQueueProcessorScope = "OutboundQueueProcessor"
+	// VersionMembershipCacheGetScope is the scope used by version membership cache
+	VersionMembershipCacheGetScope = "VersionMembershipCacheGet"
+	// VersionMembershipCachePutScope is the scope used by version membership cache
+	VersionMembershipCachePutScope = "VersionMembershipCachePut"
 )
 
 // Matching Scope

common/worker_versioning/version_membership_cache.go

@@ -0,0 +1,108 @@
+//nolint:staticcheck
+package worker_versioning
+
+import (
+	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/server/common/cache"
+	"go.temporal.io/server/common/metrics"
+)
+
+// VersionMembershipCache is used to cache results of Matching's CheckTaskQueueVersionMembership
+// calls (used internally by the worker versioning pinned override validation).
+//
+// Implementations are expected to be safe for concurrent use.
+type (
+	VersionMembershipCache interface {
+		// Get returns (isMember, ok). ok=false means there was no cached value.
+		Get(
+			namespaceID string,
+			taskQueue string,
+			taskQueueType enumspb.TaskQueueType,
+			deploymentName string,
+			buildID string,
+		) (isMember bool, ok bool)
+
+		Put(
+			namespaceID string,
+			taskQueue string,
+			taskQueueType enumspb.TaskQueueType,
+			deploymentName string,
+			buildID string,
+			isMember bool,
+		)
+	}
+
+	versionMembershipCacheKey struct {
+		namespaceID    string
+		taskQueue      string
+		taskQueueType  enumspb.TaskQueueType
+		deploymentName string
+		buildID        string
+	}
+
+	VersionMembershipCacheImpl struct {
+		cache.Cache
+		metricsHandler metrics.Handler
+	}
+)
+
+// NewVersionMembershipCache wraps the provided cache with a typed API and metrics.
+func NewVersionMembershipCache(c cache.Cache, metricsHandler metrics.Handler) VersionMembershipCache {
+	h := metricsHandler.WithTags(metrics.CacheTypeTag(metrics.VersionMembershipCacheTypeTagValue))
+	return &VersionMembershipCacheImpl{
+		Cache:          c,
+		metricsHandler: h,
+	}
+}
+
+func (c *VersionMembershipCacheImpl) Get(
+	namespaceID string,
+	taskQueue string,
+	taskQueueType enumspb.TaskQueueType,
+	deploymentName string,
+	buildID string,
+) (isMember bool, ok bool) {
+	handler := c.metricsHandler.WithTags(metrics.OperationTag(metrics.VersionMembershipCacheGetScope), metrics.NamespaceIDTag(namespaceID))
+	metrics.CacheRequests.With(handler).Record(1)
+
+	key := versionMembershipCacheKey{
+		namespaceID:    namespaceID,
+		taskQueue:      taskQueue,
+		taskQueueType:  taskQueueType,
+		deploymentName: deploymentName,
+		buildID:        buildID,
+	}
+	v := c.Cache.Get(key)
+	if v == nil {
+		metrics.CacheMissCounter.With(handler).Record(1)
+		return false, false
+	}
+	isMember, ok = v.(bool)
+	if !ok {
+		// Unexpected type: treat as miss to avoid false positives.
+		metrics.CacheMissCounter.With(handler).Record(1)
+		return false, false
+	}
+	return isMember, true
+}
+
+func (c *VersionMembershipCacheImpl) Put(
+	namespaceID string,
+	taskQueue string,
+	taskQueueType enumspb.TaskQueueType,
+	deploymentName string,
+	buildID string,
+	isMember bool,
+) {
+	handler := c.metricsHandler.WithTags(metrics.OperationTag(metrics.VersionMembershipCachePutScope), metrics.NamespaceIDTag(namespaceID))
+	metrics.CacheRequests.With(handler).Record(1)
+
+	key := versionMembershipCacheKey{
+		namespaceID:    namespaceID,
+		taskQueue:      taskQueue,
+		taskQueueType:  taskQueueType,
+		deploymentName: deploymentName,
+		buildID:        buildID,
+	}
+	c.Cache.Put(key, isMember)
+}

common/worker_versioning/worker_versioning.go

@@ -20,7 +20,6 @@ import (
 	"go.temporal.io/server/api/matchingservice/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	taskqueuespb "go.temporal.io/server/api/taskqueue/v1"
-	"go.temporal.io/server/common/cache"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/persistence/visibility/manager"
 	"go.temporal.io/server/common/resource"
@@ -530,28 +529,25 @@ func ExtractVersioningBehaviorFromOverride(override *workflowpb.VersioningOverri
 func validatePinnedVersionInTaskQueue(ctx context.Context,
 	pinnedVersion *deploymentpb.WorkerDeploymentVersion,
 	matchingClient resource.MatchingClient,
-	versionMembershipCache cache.Cache,
+	versionMembershipCache VersionMembershipCache,
 	tq string,
 	tqType enumspb.TaskQueueType,
 	namespaceID string) error {
 
 	// Check if we have recently queried matching to validate if this version exists in the task queue.
-	key := versionMembershipCacheKey{
-		namespaceID:    namespaceID,
-		taskQueue:      tq,
-		taskQueueType:  tqType,
-		deploymentName: pinnedVersion.DeploymentName,
-		buildID:        pinnedVersion.BuildId,
-	}
-	if cached := versionMembershipCache.Get(key); cached != nil {
-		if isMember, ok := cached.(bool); ok {
-			if isMember {
-				return nil
-			}
-			return serviceerror.NewFailedPrecondition(
-				"Pinned version is not present in the task queue",
-			)
+	if isMember, ok := versionMembershipCache.Get(
+		namespaceID,
+		tq,
+		tqType,
+		pinnedVersion.DeploymentName,
+		pinnedVersion.BuildId,
+	); ok {
+		if isMember {
+			return nil
 		}
+		return serviceerror.NewFailedPrecondition(
+			"Pinned version is not present in the task queue",
+		)
 	}
 
 	resp, err := matchingClient.CheckTaskQueueVersionMembership(ctx, &matchingservice.CheckTaskQueueVersionMembershipRequest{
@@ -565,7 +561,14 @@ func validatePinnedVersionInTaskQueue(ctx context.Context,
 	}
 
 	// Add result to cache
-	versionMembershipCache.Put(key, resp.GetIsMember())
+	versionMembershipCache.Put(
+		namespaceID,
+		tq,
+		tqType,
+		pinnedVersion.DeploymentName,
+		pinnedVersion.BuildId,
+		resp.GetIsMember(),
+	)
 	if !resp.GetIsMember() {
 		return serviceerror.NewFailedPrecondition(
 			"Pinned version is not present in the task queue",
@@ -574,18 +577,10 @@ func validatePinnedVersionInTaskQueue(ctx context.Context,
 	return nil
 }
 
-type versionMembershipCacheKey struct {
-	namespaceID    string
-	taskQueue      string
-	taskQueueType  enumspb.TaskQueueType
-	deploymentName string
-	buildID        string
-}
-
 func ValidateVersioningOverride(ctx context.Context,
 	override *workflowpb.VersioningOverride,
 	matchingClient resource.MatchingClient,
-	versionMembershipCache cache.Cache,
+	versionMembershipCache VersionMembershipCache,
 	tq string,
 	tqType enumspb.TaskQueueType,
 	namespaceID string) error {