1526
// later because it's not possible normally and user has to pass IgnoreMissingTaskQueues or AllowNoPollers for it to
1527
// happen, or the task queue needs to be a task queue that is added in this version.
1528
>
func (d *VersionWorkflowRunner) syncRegisteredTaskQueueAsync(ctx workflow.Context, args *deploymentspb.RegisterWorkerInVersionArgs) {
version_workflow.go
1529
>
startTime := workflow.Now(ctx)
1530
>
1531
>
versionData := &deploymentspb.WorkerDeploymentVersionData{
1532
>
Status: d.VersionState.Status,
1533
>
RevisionNumber: d.GetVersionState().GetRevisionNumber(),
1534
>
UpdateTime: timestamppb.New(workflow.Now(ctx)),
1535
>
}
1536
>
1537
>
batch := []*deploymentspb.SyncDeploymentVersionUserDataRequest_SyncUserData{
1538
>
{Name: args.GetTaskQueueName(), Types: []enumspb.TaskQueueType{args.GetTaskQueueType()}},
1539
>
}
1540
>
1541
>
// This must be done in the main goroutine otherwise the wf may CaN before getting to it.
1542
>
d.asyncPropagationsInProgress++
1543
>
1544
>
// Start async propagation - DON'T WAIT
1545
>
workflow.Go(ctx, func(gCtx workflow.Context) {
1546
>
// We only sync to the root partition and don't wait for all partitions propagation, this is because
1547
>
// the task queue partition initiating the registration itself is responsible to block until it sees the version in user data.
1548
>
d.executePropagationBatch(gCtx, batch, args.GetRoutingConfig(), versionData)
1549
>
1550
>
d.metrics.Timer(metrics.VersioningDataPropagationLatency.Name()).Record(workflow.Now(ctx).Sub(startTime))
1551
>
// Decrement counter when propagation completes
1552
>
d.asyncPropagationsInProgress--
1553
>
})
1554
}