Vihren Atlas

matching_engine.go ×23

Code frontier

unlabeled · c_02f044c7c615

36 tests · 5380 LOC · 192 files · introduces 0 tests · 111 LOC · 3 files

Introduces — evidence that enters the hierarchy at this concept

Code
37 ranges111 lines · 3 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
1079 ranges5380 lines · 192 files · Browse complete extent
All tests (intent)
36 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

3 files ranked by introduced lines: 111 introduced LOC across 37 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/matching/matching_engine.go 71 introduced LOC · 23 ranges

Open complete file

2199 ctx context.Context,
2200 req *matchingservice.SyncDeploymentUserDataRequest,
2201 > ) (*matchingservice.SyncDeploymentUserDataResponse, error) { matching_engine.go
2202 > taskQueueFamily, err := tqid.NewTaskQueueFamily(req.NamespaceId, req.GetTaskQueue())
2203 > applyUpdatesToRoutingConfig := false
2204 >
2205 > if err != nil {
2206 return nil, err
2207 }
2208 > if req.GetOperation() == nil && req.GetDeploymentName() == "" { matching_engine.go
2209 return nil, errMissingDeploymentVersion
2210 }
2211
2212 > tqMgr, _, err := e.getTaskQueuePartitionManager(ctx, taskQueueFamily.TaskQueue(enumspb.TASK_QUEUE_TYPE_WORKFLOW).RootPartition(), true, loadCauseOtherWrite) matching_engine.go
2213 > if err != nil {
2214 return nil, err
2215 }
2216
2217 > updateOptions := UserDataUpdateOptions{Source: "SyncDeploymentUserData"} matching_engine.go
2218 >
2219 > version, err := tqMgr.GetUserDataManager().UpdateUserData(ctx, updateOptions, func(data *persistencespb.TaskQueueUserData) (*persistencespb.TaskQueueUserData, bool, error) {
2220 > clk := data.GetClock()
2221 > if clk == nil {
2222 > clk = hlc.Zero(e.clusterMeta.GetClusterID())
2223 > }
2224 > now := hlc.Next(clk, e.timeSource)
2225 > // clone the whole thing so we can just mutate
2226 > data = common.CloneProto(data)
2227 >
2228 > // fill in enough structure so that we can set/append the new deployment data
2229 > if data == nil {
2230 data = &persistencespb.TaskQueueUserData{}
2231 }
2232 > if data.PerType == nil { matching_engine.go
2233 data.PerType = make(map[int32]*persistencespb.TaskQueueTypeUserData)
2234 }
2235
2236 > changed := false matching_engine.go
2237 > for _, t := range req.TaskQueueTypes {
2238 > if data.PerType[int32(t)] == nil {
2239 data.PerType[int32(t)] = &persistencespb.TaskQueueTypeUserData{}
2240 }
2241 > if data.PerType[int32(t)].DeploymentData == nil { matching_engine.go
2242 data.PerType[int32(t)].DeploymentData = &persistencespb.DeploymentData{}
2243 }
2244
2245 // set/append the new data
2246 > deploymentData := data.PerType[int32(t)].DeploymentData matching_engine.go
2247 >
2248 > //nolint:staticcheck // SA1019
2249 > if vd := req.GetUpdateVersionData(); vd != nil {
2250 // [cleanup-public-preview-versioning]
2251 if vd.GetVersion() == nil { // unversioned ramp
2287 clearVersionFromRoutingConfig(workerDeploymentData, nil, vd)
2288 }
2289 > } else if v := req.GetForgetVersion(); v != nil { matching_engine.go
2290 // Go through the new and old deployment data format for this deployment and remove the version if present.
2291 workerDeploymentData := deploymentData.GetDeploymentsData()[v.GetDeploymentName()]
2300 changed = true
2301 }
2302 > } else { matching_engine.go
2303 >
2304 > // Only initialize DeploymentsData if we're using the new format
2305 > if deploymentData.GetDeploymentsData() == nil {
2306 > deploymentData.DeploymentsData = make(map[string]*persistencespb.WorkerDeploymentData)
2307 > }
2308 > if deploymentData.GetDeploymentsData()[req.GetDeploymentName()] == nil {
2309 > deploymentData.GetDeploymentsData()[req.GetDeploymentName()] = &persistencespb.WorkerDeploymentData{}
2310 > }
2311
2312 > rc := req.GetUpdateRoutingConfig() matching_engine.go
2313 > tqWorkerDeploymentData := deploymentData.GetDeploymentsData()[req.GetDeploymentName()]
2314 >
2315 > ignoreRevCheck, _ := testhooks.Get(e.testHooks, testhooks.MatchingIgnoreRoutingConfigRevisionCheck, namespace.ID(req.NamespaceId))
2316 > if ignoreRevCheck || rc.GetRevisionNumber() > tqWorkerDeploymentData.GetRoutingConfig().GetRevisionNumber() {
2317 changed = true
2318 // Update routing config when newer or equal revision is provided
2321 }
2322
2323 > if tqWorkerDeploymentData.Versions == nil { matching_engine.go
2324 > tqWorkerDeploymentData.Versions = make(map[string]*deploymentspb.WorkerDeploymentVersionData)
2325 > }
2326 > for buildID, versionData := range req.GetUpsertVersionsData() {
2327 existing := tqWorkerDeploymentData.Versions[buildID]
2328 // Skip if existing version data has a higher revision number to avoid stale writes.
2349 }
2350
2351 > if removed := removeDeploymentVersions( matching_engine.go
2352 > deploymentData,
2353 > req.GetDeploymentName(),
2354 > tqWorkerDeploymentData,
2355 > req.GetForgetVersions(),
2356 > /* removeOldFormat */ true,
2357 > ); removed {
2358 changed = true
2359 }
2384 */
2385
2386 > if applyUpdatesToRoutingConfig { matching_engine.go
2387 migrateOldFormatVersions(
2388 deploymentData,
2392 }
2393
2394 > if worker_versioning.CleanupOldDeletedVersions(tqWorkerDeploymentData, e.config.MaxVersionsInTaskQueue(tqMgr.Namespace().Name().String())) { matching_engine.go
2395 changed = true
2396 }
2397 }
2398 }
2399 > if !changed { matching_engine.go
2400 return nil, false, errUserDataUnmodified
2401 }
2402
2403 > data.Clock = now matching_engine.go
2404 > return data, true, nil
2405 })
2406 > if err != nil { matching_engine.go
2407 return nil, err
2408 }
2409 > return &matchingservice.SyncDeploymentUserDataResponse{Version: version, RoutingConfigChanged: applyUpdatesToRoutingConfig}, nil matching_engine.go
2410 }
2411
3859 buildIDs []string,
3860 removeOldFormat bool,
3861 > ) bool { matching_engine.go
3862 > if workerDeploymentData == nil && !removeOldFormat {
3863 return false
3864 }
3865 > changed := false matching_engine.go
3866 > deletedInNew := false
3867 >
3868 > for _, buildID := range buildIDs {
3869 if _, exists := workerDeploymentData.GetVersions()[buildID]; exists {
3870 delete(workerDeploymentData.GetVersions(), buildID)
3887
3888 // Only remove the deployment entry if versions were actually deleted from the new-format map.
3889 > if workerDeploymentData != nil && deletedInNew && len(workerDeploymentData.GetVersions()) == 0 { matching_engine.go
3890 delete(deploymentData.GetDeploymentsData(), deploymentName)
3891 }
3892 > return changed matching_engine.go
3893 }
3894
go.temporal.io/server/api/matchingservice/v1/request_response.pb.go 32 introduced LOC · 10 ranges

Open complete file

2988 }
2989
2990 > func (x *SyncDeploymentUserDataRequest) GetTaskQueue() string { request_response.pb.go
2991 > if x != nil {
2992 > return x.TaskQueue
2993 > }
2994 return ""
2995 }
2996
2997 > func (x *SyncDeploymentUserDataRequest) GetDeploymentName() string { request_response.pb.go
2998 > if x != nil {
2999 > return x.DeploymentName
3000 > }
3001 return ""
3002 }
3009 }
3010
3011 > func (x *SyncDeploymentUserDataRequest) GetOperation() isSyncDeploymentUserDataRequest_Operation { request_response.pb.go
3012 > if x != nil {
3013 > return x.Operation
3014 > }
3015 return nil
3016 }
3017
3018 // Deprecated: Marked as deprecated in temporal/server/api/matchingservice/v1/request_response.proto.
3019 > func (x *SyncDeploymentUserDataRequest) GetUpdateVersionData() *v110.DeploymentVersionData { request_response.pb.go
3020 > if x != nil {
3021 > if x, ok := x.Operation.(*SyncDeploymentUserDataRequest_UpdateVersionData); ok {
3022 return x.UpdateVersionData
3023 }
3024 }
3025 > return nil request_response.pb.go
3026 }
3027
3028 // Deprecated: Marked as deprecated in temporal/server/api/matchingservice/v1/request_response.proto.
3029 > func (x *SyncDeploymentUserDataRequest) GetForgetVersion() *v110.WorkerDeploymentVersion { request_response.pb.go
3030 > if x != nil {
3031 > if x, ok := x.Operation.(*SyncDeploymentUserDataRequest_ForgetVersion); ok {
3032 return x.ForgetVersion
3033 }
3034 }
3035 > return nil request_response.pb.go
3036 }
3037
3038 > func (x *SyncDeploymentUserDataRequest) GetUpdateRoutingConfig() *v112.RoutingConfig { request_response.pb.go
3039 > if x != nil {
3040 > return x.UpdateRoutingConfig
3041 > }
3042 return nil
3043 }
3044
3045 > func (x *SyncDeploymentUserDataRequest) GetUpsertVersionsData() map[string]*v110.WorkerDeploymentVersionData { request_response.pb.go
3046 > if x != nil {
3047 > return x.UpsertVersionsData
3048 > }
3049 return nil
3050 }
3051
3052 > func (x *SyncDeploymentUserDataRequest) GetForgetVersions() []string { request_response.pb.go
3053 > if x != nil {
3054 > return x.ForgetVersions
3055 > }
3056 return nil
3057 }
go.temporal.io/server/api/persistence/v1/task_queues.pb.go 8 introduced LOC · 4 ranges

Open complete file

486 func (*DeploymentData) ProtoMessage() {}
487
488 > func (x *DeploymentData) ProtoReflect() protoreflect.Message { task_queues.pb.go
489 > mi := &file_temporal_server_api_persistence_v1_task_queues_proto_msgTypes[5]
490 > if x != nil {
491 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
492 if ms.LoadMessageInfo() == nil {
551 func (*WorkerDeploymentData) ProtoMessage() {}
552
553 > func (x *WorkerDeploymentData) ProtoReflect() protoreflect.Message { task_queues.pb.go
554 > mi := &file_temporal_server_api_persistence_v1_task_queues_proto_msgTypes[6]
555 > if x != nil {
556 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
557 if ms.LoadMessageInfo() == nil {

Direct hierarchy

Parents

Children