worker-versioning GA: revision number to handle async workflow inconsistencies. #8553
Conversation
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
… if a version is ramping or not
…ping only when no other version in any deployment is current/ramping
…heck instead of FindDeploymentVersion
service/matching/matching_engine.go
Outdated
| dst := make([]*deploymentspb.DeploymentVersionData, 0, len(oldVersions)) | ||
| for _, dv := range oldVersions { | ||
| if dv.GetVersion().GetDeploymentName() != req.GetDeploymentName() { | ||
| dst = append(dst, dv) |
There was a problem hiding this comment.
need to also add it to tqWorkerDeploymentData.Versions
| } else if vd := req.GetUpdateVersionData(); vd != nil { | ||
| // [cleanup-public-preview-versioning] | ||
| if vd.GetVersion() == nil { // unversioned ramp | ||
| if deploymentData.GetUnversionedRampData().GetRoutingUpdateTime().AsTime().After(vd.GetRoutingUpdateTime().AsTime()) { |
There was a problem hiding this comment.
first send the deployment name in the old format then check new format for the deployment name and unset the ramping version and its timestamps
| } | ||
| changed = true | ||
| // only update if the timestamp is more recent | ||
| deploymentData.Versions[idx] = vd |
There was a problem hiding this comment.
after doing old stuff, check the new format for the given deployment name and depending on this version being ramping or current, clean up the fields in new format routing config.
service/matching/matching_engine.go
Outdated
| deploymentData.Versions = append(deploymentData.Versions, vd) | ||
| } | ||
| } else if v := req.GetForgetVersion(); v != nil { | ||
| // TODO (Shivam): Remove versions from both places. |
There was a problem hiding this comment.
yup need to remove from both places
service/matching/matching_engine.go
Outdated
| changed = true | ||
| } | ||
| for _, buildID := range req.GetForgetVersions() { | ||
| delete(tqWorkerDeploymentData.Versions, buildID) |
There was a problem hiding this comment.
need to remove from old format too.
Also, when all versions of the deployment are removed, clean up the whole deployment data from the upper map.
What changed?
Why?
How did you test it?
Potential risks
Note
Introduce revision-number mechanics for task dispatch and a new per-deployment routing config/user data schema to handle eventual consistency in worker versioning.
TaskVersionDirectiveand propagate astask_dispatch_revision_numberthrough Matching → History for WFT/AT start requests.StartDeploymentTransitionnow accepts and records it; directive builders include revision.taskDispatchRevisionNumberfrom routing config vs directive and selects target queue accordingly (gated by dynamic configUseRevisionNumberForWorkerVersioning).DeploymentVersionDatadeprecated; addWorkerDeploymentVersionData.SyncDeploymentUserDataRequestwithdeployment_name,update_routing_config,upsert_versions_data,forget_versions; response addsrouting_config_changed.DeploymentData.deployments_data(map) andWorkerDeploymentData{routing_config, versions}.FindTargetDeploymentVersionAndRevisionNumberForWorkflowIDand enhancedCalculateTaskQueueVersioningInfoto reconcile old/new data and return revision numbers.system.useRevisionNumberForWorkerVersioningand plumb into History/Matching.Written by Cursor Bugbot for commit f5a5f6e. This will update automatically on new commits. Configure here.