mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
## 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]>
66 lines
2.1 KiB
Go
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
|
|
}()
|
|
}
|