1340
// syncTaskQueuesAsync must be called within the lock. It first increments the version revision number and then
1341
// starts async propagation of version data (and routing info if given) to all task queues.
1342
>
func (d *VersionWorkflowRunner) syncTaskQueuesAsync(ctx workflow.Context, routingConfig *deploymentpb.RoutingConfig, versionDataChanged bool) {
version_workflow.go
1343
>
startTime := workflow.Now(ctx)
1344
>
1345
>
withRevisionNumber := d.hasMinVersion(VersionDataRevisionNumber)
1346
>
if withRevisionNumber && versionDataChanged {
1347
d.GetVersionState().RevisionNumber++
1348
}
1349
1351
>
Status: d.VersionState.Status,
1352
>
RevisionNumber: d.GetVersionState().GetRevisionNumber(),
1353
>
UpdateTime: timestamppb.New(workflow.Now(ctx)),
1354
>
Deleted: d.deleteVersion,
1355
>
}
1356
>
1357
>
// Batches must be calculated within the lock otherwise previous update might be called on future task queues unintentionally.
1358
>
batches := d.batchTaskQueuesForSync()
1359
>
1360
>
// Increment counter to prevent CaN while propagation is in progress.
1361
>
// This must be done in the main goroutine otherwise the wf may CaN before getting to it.
1362
>
d.asyncPropagationsInProgress++
1363
>
1364
>
// Start async propagation - DON'T WAIT
1365
>
workflow.Go(ctx, func(gCtx workflow.Context) {
1366
>
d.executeAndTrackAsyncPropagation(gCtx, batches, routingConfig, versionData)
1367
>
1368
>
if routingConfig != nil && withRevisionNumber {
1369
// Signal deployment workflow that routing config propagation completed
1370
d.signalPropagationComplete(gCtx, routingConfig.GetRevisionNumber())
1371
}
1372
1373
>
d.metrics.Timer(metrics.VersioningDataPropagationLatency.Name()).Record(workflow.Now(ctx).Sub(startTime))
version_workflow.go
1374
>
// Decrement counter when propagation completes
1375
>
d.asyncPropagationsInProgress--
1376
})
1377
}