mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
1e31c76ed1
## 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]>
107 lines
3.6 KiB
Go
107 lines
3.6 KiB
Go
package signalwithstartworkflow
|
|
|
|
import (
|
|
"context"
|
|
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
"go.temporal.io/api/serviceerror"
|
|
"go.temporal.io/server/api/historyservice/v1"
|
|
"go.temporal.io/server/api/matchingservice/v1"
|
|
"go.temporal.io/server/common/definition"
|
|
"go.temporal.io/server/common/enums"
|
|
"go.temporal.io/server/common/locks"
|
|
"go.temporal.io/server/common/metrics"
|
|
"go.temporal.io/server/common/namespace"
|
|
"go.temporal.io/server/common/worker_versioning"
|
|
"go.temporal.io/server/service/history/api"
|
|
historyi "go.temporal.io/server/service/history/interfaces"
|
|
)
|
|
|
|
func Invoke(
|
|
ctx context.Context,
|
|
signalWithStartRequest *historyservice.SignalWithStartWorkflowExecutionRequest,
|
|
shard historyi.ShardContext,
|
|
workflowConsistencyChecker api.WorkflowConsistencyChecker,
|
|
matchingClient matchingservice.MatchingServiceClient,
|
|
versionMembershipCache worker_versioning.VersionMembershipCache,
|
|
reactivationSignalCache worker_versioning.ReactivationSignalCache,
|
|
reactivationSignaler api.VersionReactivationSignalerFn,
|
|
) (_ *historyservice.SignalWithStartWorkflowExecutionResponse, retError error) {
|
|
namespaceEntry, err := api.GetActiveNamespace(shard, namespace.ID(signalWithStartRequest.GetNamespaceId()), signalWithStartRequest.SignalWithStartRequest.WorkflowId)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
namespaceID := namespaceEntry.ID()
|
|
|
|
var currentWorkflowLease api.WorkflowLease
|
|
currentWorkflowLease, err = workflowConsistencyChecker.GetWorkflowLease(
|
|
ctx,
|
|
nil,
|
|
definition.NewWorkflowKey(
|
|
string(namespaceID),
|
|
signalWithStartRequest.SignalWithStartRequest.WorkflowId,
|
|
"",
|
|
),
|
|
locks.PriorityHigh,
|
|
)
|
|
switch err.(type) {
|
|
case nil:
|
|
defer func() { currentWorkflowLease.GetReleaseFn()(retError) }()
|
|
case *serviceerror.NotFound:
|
|
currentWorkflowLease = nil
|
|
default:
|
|
return nil, err
|
|
}
|
|
|
|
// TODO: remove this call in 1.25
|
|
enums.SetDefaultWorkflowIdConflictPolicy(
|
|
&signalWithStartRequest.SignalWithStartRequest.WorkflowIdConflictPolicy,
|
|
enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING)
|
|
|
|
api.MigrateWorkflowIdReusePolicyForRunningWorkflow(
|
|
&signalWithStartRequest.SignalWithStartRequest.WorkflowIdReusePolicy,
|
|
&signalWithStartRequest.SignalWithStartRequest.WorkflowIdConflictPolicy)
|
|
|
|
startRequest := ConvertToStartRequest(
|
|
namespaceID,
|
|
signalWithStartRequest.SignalWithStartRequest,
|
|
shard.GetTimeSource().Now(),
|
|
)
|
|
request := startRequest.StartRequest
|
|
|
|
api.OverrideStartWorkflowExecutionRequest(request, metrics.HistorySignalWithStartWorkflowExecutionScope, shard, shard.GetMetricsHandler())
|
|
|
|
err = api.ValidateStartWorkflowExecutionRequest(ctx, request, shard, namespaceEntry, "SignalWithStartWorkflowExecution")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Validation for versioning override, if any.
|
|
err = worker_versioning.ValidateVersioningOverride(ctx, request.GetVersioningOverride(), matchingClient, versionMembershipCache, request.GetTaskQueue().GetName(), enumspb.TASK_QUEUE_TYPE_WORKFLOW, namespaceID.String())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
runID, started, err := SignalWithStartWorkflow(
|
|
ctx,
|
|
shard,
|
|
namespaceEntry,
|
|
currentWorkflowLease,
|
|
startRequest,
|
|
signalWithStartRequest.SignalWithStartRequest,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Notify version workflow if we're starting a new workflow pinned to a potentially drained version
|
|
if started {
|
|
api.ReactivateVersionWorkflowIfPinned(ctx, namespaceEntry, request.GetVersioningOverride(), reactivationSignalCache, reactivationSignaler, shard.GetConfig().EnableVersionReactivationSignals())
|
|
}
|
|
|
|
return &historyservice.SignalWithStartWorkflowExecutionResponse{
|
|
RunId: runID,
|
|
Started: started,
|
|
}, nil
|
|
}
|