639
}
640
641
>
d.State.Versions[args.Version] = &deploymentspb.WorkerDeploymentVersionSummary{
workflow.go
642
>
Version: args.Version,
643
>
CreateTime: args.CreateTime,
644
>
Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE,
645
>
}
646
>
d.metrics.Counter(metrics.WorkerDeploymentVersionCreated.Name()).Inc(1)
647
>
return nil
648
}
649
650
>
func (d *WorkflowRunner) handleRegisterWorker(ctx workflow.Context, args *deploymentspb.RegisterWorkerInWorkerDeploymentArgs) error {
workflow.go
651
>
// TODO: there is a small race condition where the deployment is just deleted and got a register update before closing itself.
652
>
// In that case, we should ideally not reject the update, but revive the workflow so that the caller does not need to retry.
653
>
// In practice this should be fine because the polls will retry and Deployment workflows are short-lived.
654
>
// Same principle applies for Version workflows, but they can be slightly more long-lived of they are handling long propagations.
655
>
// Hence, the revive logic is implemented in Version workflow but not here yet.
656
>
if err := d.ensureNotDeleted(); err != nil {
657
return err
658
}
659
660
// use lock to enforce only one update at a time
662
>
if err != nil {
663
d.logger.Error("Could not acquire workflow lock")
664
return serviceerror.NewDeadlineExceeded("Could not acquire workflow lock")
665
}
667
>
// Even if the update doesn't change the state we mark it as dirty because of created history events.
668
>
d.setStateChanged()
669
>
d.lock.Unlock()
670
>
}()
671
672
>
version := worker_versioning.WorkerDeploymentVersionToStringV31(args.Version)
workflow.go
673
>
674
>
// Add version to local state of the workflow, if not already present.
675
>
err = d.addVersionToWorkerDeployment(ctx, &deploymentspb.AddVersionUpdateArgs{
676
>
Version: version,
677
>
CreateTime: timestamppb.New(workflow.Now(ctx)),
678
>
})
679
>
if err != nil {
680
return err
681
}
682
>
var routingConfigToSync *deploymentpb.RoutingConfig
workflow.go
683
>
if d.hasMinVersion(AsyncSetCurrentAndRamping) {
684
routingConfigToSync = d.GetState().GetRoutingConfig()
685
}
686
687
// Register task-queue worker in version workflow.
688
>
activityCtx := workflow.WithActivityOptions(ctx, defaultActivityOptions)
workflow.go
689
>
err = workflow.ExecuteActivity(activityCtx, d.a.RegisterWorkerInVersion, &deploymentspb.RegisterWorkerInVersionArgs{
690
>
TaskQueueName: args.TaskQueueName,
691
>
TaskQueueType: args.TaskQueueType,
692
>
MaxTaskQueues: args.MaxTaskQueues,
693
>
Version: version,
694
>
RoutingConfig: routingConfigToSync,
695
>
}).Get(ctx, nil)
696
>
if err != nil {
697
var appError *temporal.ApplicationError
698
if errors.As(err, &appError) {