Files
tasks/service/history/api/worker_versioning_util.go
1e31c76ed1 Versioning cherry-pick for OSS v1.30.2 (#9543)
## What changed?
Cherry-pick versioning PRs
- #9168
  - Cache for system protection
- #9262 
  - Cache for system protection
- #9239 
- Critical PR to enable sending `TargetVersionChanged` flag for
Upgrade-on-CaN feature
- #9147 
- Tracks version drainage properly when version receives workflows via
`VersioningOverride`. Needed for automated worker controllers to
correctly scale versioned workers that received workflows via
`VersioningOverride`.
- #9300 
- Needed for `approximate_backlog_count` metric to track Current and
Ramping version tasks correctly
- #9316 
- Needed for `approximate_backlog_count` metric to track Current and
Ramping version tasks correctly
- #8957 
- Contains minor metric improvement. Included because it adds a test
harness that is used in the two metrics PRs above
- #9250 
  - Bug fix of task rescheduling edge case during AutoUpgrade Transition

## Why?
For OSS v1.30.2

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





---

> [!NOTE]
> **Medium Risk**
> Touches history/matching worker-versioning paths, adding new caches
and changing workflow task/start handling and backlog metric emission;
incorrect caching or signaling could affect dispatch/upgrade behavior
and observability.
> 
> **Overview**
> Adds new worker-versioning protections and upgrade signaling: workflow
task started events now persist a
`workflow_task_target_worker_deployment_version_changed` flag (and emit
a new `workflow_target_version_changed_count` metric) under a new
`EnableSendTargetVersionChanged` dynamic config.
> 
> Introduces two new caches with metrics and dynamic config knobs: a
`RoutingInfoCache` to avoid repeated `GetTaskQueueUserData` lookups
during activity start/transition logic, and a `ReactivationSignalCache`
plus `EnableVersionReactivationSignals` to dedupe and asynchronously
send “reactivation” signals when workflows are pinned (via
start/signal-with-start/reset/update-options) to potentially
drained/inactive worker versions.
> 
> Extends matching backlog metrics to support version-attributed
reporting by adding `BacklogMetricsEmitInterval` and switching queue DB
emission to *physical* backlog gauges (`physical_approximate_backlog_*`)
when attribution is enabled, while keeping legacy gauges when disabled.
> 
> Adds frontend scaffolding for a new visibility RPC `CountSchedules`
(client plumbing, interception/metadata/quota wiring) but leaves the
frontend handler unimplemented, and bumps `go.temporal.io/api` to
`v1.62.2`.
> 
> <sup>Written by [Cursor
Bugbot]() for commit
cb8ae14c86666654c51dd66ae91cb225722e1ba7. This will update automatically
on new commits. Configure
[here]().</sup>

---------

Co-authored-by: Shahab Tajik <[email protected]>
Co-authored-by: Shivam <[email protected]>
2026-03-17 17:19:47 -05:00

66 lines
2.1 KiB
Go

package api
import (
"context"
workflowpb "go.temporal.io/api/workflow/v1"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/common/worker_versioning"
)
// VersionReactivationSignalerFn is a function type for sending reactivation signals to version workflows.
// This abstraction allows the history API layer to use the deployment client without importing it directly,
// avoiding import cycles between history/api and worker/workerdeployment packages.
type VersionReactivationSignalerFn func(
ctx context.Context,
namespaceEntry *namespace.Namespace,
deploymentName, buildID string,
) error
// ReactivateVersionWorkflowIfPinned sends a reactivation signal to the version workflow
// when workflows are pinned to a potentially DRAINED/INACTIVE version. It also deduplicates
// signals within the cache TTL window.
// This is a fire-and-forget operation - the signal is sent asynchronously and errors are
// logged by the signaler implementation.
//
//nolint:revive,errcheck
func ReactivateVersionWorkflowIfPinned(
ctx context.Context,
namespaceEntry *namespace.Namespace,
override *workflowpb.VersioningOverride,
signalCache worker_versioning.ReactivationSignalCache,
signaler VersionReactivationSignalerFn,
enabled bool,
) {
// Check if signals are enabled globally
if !enabled {
return
}
// Only process if we're pinning to a specific version
if !worker_versioning.OverrideIsPinned(override) {
return
}
pinnedVersion := worker_versioning.GetOverridePinnedVersion(override)
if pinnedVersion == nil {
return
}
// Check cache - skip if signal was recently sent
if signalCache != nil && !signalCache.ShouldSendSignal(
namespaceEntry.ID().String(),
pinnedVersion.GetDeploymentName(),
pinnedVersion.GetBuildId(),
) {
return
}
// Send the signal asynchronously to avoid adding latency to the caller's request.
// Errors are logged by the signaler implementation (e.g. via convertAndRecordError). However,
// errors are not propagated to the caller as this is a fire-and-forget operation.
go func() {
signaler(context.Background(), namespaceEntry, pinnedVersion.GetDeploymentName(), pinnedVersion.GetBuildId()) //nolint:errcheck
}()
}