Vihren Atlas

workflow.go ×15

Code frontier

unlabeled · c_856e87de6704

79 tests · 3029 LOC · 133 files · introduces 0 tests · 82 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
17 ranges82 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
473 ranges3029 lines · 133 files · Browse complete extent
All tests (intent)
79 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.

2 files ranked by introduced lines: 82 introduced LOC across 17 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/worker/workerdeployment/workflow.go 68 introduced LOC · 15 ranges

Open complete file

1251 }
1252
1253 > func (d *WorkflowRunner) validateStateBeforeAcceptingSetCurrent(args *deploymentspb.SetCurrentVersionArgs) error { workflow.go
1254 > //nolint:staticcheck // SA1019: worker versioning v0.31
1255 > if d.State.GetRoutingConfig().GetCurrentVersion() == args.Version && d.State.GetLastModifierIdentity() == args.Identity {
1256 return temporal.NewApplicationError("no change", errNoChangeType, d.State.ConflictToken)
1257 }
1258 > if args.ConflictToken != nil && !bytes.Equal(args.ConflictToken, d.State.ConflictToken) { workflow.go
1259 return temporal.NewApplicationError("conflict token mismatch", errFailedPrecondition)
1260 }
1261 > if _, ok := d.State.Versions[args.Version]; !ok && workflow.go
1262 > args.Version != worker_versioning.UnversionedVersionId &&
1263 > !args.GetAllowNoPollers() {
1264 d.logger.Info("version not found in deployment")
1265 return temporal.NewApplicationError(fmt.Sprintf("version %s not found in deployment", args.Version), errVersionNotFound)
1266 }
1267 > if d.State.ManagerIdentity != "" && d.State.ManagerIdentity != args.Identity { workflow.go
1268 return serviceerror.NewFailedPrecondition(fmt.Sprintf(ErrManagerIdentityMismatch, d.State.ManagerIdentity, args.Identity))
1269 }
1270 > return nil workflow.go
1271 }
1272
1280
1281 //nolint:staticcheck // deprecated stuff will be cleaned
1282 > func (d *WorkflowRunner) handleSetCurrent(ctx workflow.Context, args *deploymentspb.SetCurrentVersionArgs) (*deploymentspb.SetCurrentVersionResponse, error) { workflow.go
1283 > if err := d.preUpdateChecks(ctx); err != nil {
1284 return nil, err
1285 }
1286
1287 // use lock to enforce only one update at a time
1288 > err := d.lock.Lock(ctx) workflow.go
1289 > if err != nil {
1290 d.logger.Error("Could not acquire workflow lock")
1291 return nil, serviceerror.NewDeadlineExceeded("Could not acquire workflow lock")
1292 }
1293 > defer func() { workflow.go
1294 > // Even if the update doesn't change the state we mark it as dirty because of created history events.
1295 > d.setStateChanged()
1296 > d.lock.Unlock()
1297 > }()
1298
1299 // Log state before update
1300 // TODO(carlydf): remove verbose logging
1301 > d.logger.Info("Starting SetCurrent update", workflow.go
1302 > //nolint:staticcheck // SA1019: worker versioning v0.31
1303 > "current_version", d.State.GetRoutingConfig().GetCurrentVersion(),
1304 > "new_version", args.Version,
1305 > "routing_config", d.State.GetRoutingConfig())
1306 >
1307 > // Validating the state before starting the SetCurrent operation. This is required due to the following reason:
1308 > // The validator accepts/rejects updates based on the state of the deployment workflow. Theoretically, two concurrent update requests
1309 > // might be accepted by the validator since the state of the workflow, at that point in time, is valid for the updates to take place. Since this update handler
1310 > // enforces sequential updates, after the first update completes, the local state of the deployment workflow will change. The second update,
1311 > // now already accepted by the validator, should now not be allowed to run since the state of the workflow is different.
1312 > err = d.validateStateBeforeAcceptingSetCurrent(args)
1313 > if err != nil {
1314 return nil, err
1315 }
1316
1317 > prevCurrentVersion := d.State.RoutingConfig.CurrentVersion workflow.go
1318 > newCurrentVersion := args.Version
1319 > updateTime := timestamppb.New(workflow.Now(ctx))
1320 >
1321 > if _, ok := d.State.Versions[args.Version]; !ok &&
1322 > args.Version != worker_versioning.UnversionedVersionId &&
1323 > args.GetAllowNoPollers() {
1324 d.logger.Info("version not found in deployment, but AllowNoPollers is true, so we will create the version")
1325 if err := d.addVersionToWorkerDeployment(ctx, &deploymentspb.AddVersionUpdateArgs{Version: newCurrentVersion, CreateTime: updateTime}); err != nil {
1339 }
1340
1341 > if !args.IgnoreMissingTaskQueues && workflow.go
1342 > prevCurrentVersion != worker_versioning.UnversionedVersionId &&
1343 > newCurrentVersion != worker_versioning.UnversionedVersionId {
1344 isMissingTaskQueues, err := d.isVersionMissingTaskQueues(ctx, prevCurrentVersion, newCurrentVersion)
1345 if err != nil {
1356 }
1357
1358 > asyncMode := d.hasMinVersion(AsyncSetCurrentAndRamping) workflow.go
1359 >
1360 > if prevCurrentVersion != newCurrentVersion || !asyncMode {
1361 > // In async mode we do not touch routing config and versions if this is not changing the
1362 > // current version (but could still come here because modifier identity is changing).
1363 >
1364 > // Build pending routing config with the updated current version
1365 > // Initialize for both sync and async modes to simplify state update logic
1366 > // Ensure RampingDeploymentVersion is populated from deprecated field if needed for backward compatibility
1367 > rampingDeploymentVersion := d.State.RoutingConfig.RampingDeploymentVersion
1368 > if rampingDeploymentVersion == nil {
1369 rampingDeploymentVersion = worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(d.State.RoutingConfig.RampingVersion)
1370 }
1371 > pendingRoutingConfig := &deploymentpb.RoutingConfig{ workflow.go
1372 > CurrentDeploymentVersion: worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(newCurrentVersion),
1373 > CurrentVersion: newCurrentVersion,
1374 > RampingDeploymentVersion: rampingDeploymentVersion,
1375 > RampingVersion: d.State.RoutingConfig.RampingVersion,
1376 > RampingVersionPercentage: d.State.RoutingConfig.RampingVersionPercentage,
1377 > CurrentVersionChangedTime: updateTime,
1378 > RampingVersionChangedTime: d.State.RoutingConfig.RampingVersionChangedTime,
1379 > RampingVersionPercentageChangedTime: d.State.RoutingConfig.RampingVersionPercentageChangedTime,
1380 > RevisionNumber: d.State.RoutingConfig.RevisionNumber,
1381 > }
1382 >
1383 > var routingConfigToSync *deploymentpb.RoutingConfig
1384 > if asyncMode {
1385 pendingRoutingConfig.RevisionNumber++
1386 // only setting it in the request if it's async mode
1389
1390 // Unset ramping if it's being promoted to current
1391 > if newCurrentVersion == d.State.RoutingConfig.RampingVersion { workflow.go
1392 pendingRoutingConfig.RampingVersion = ""
1393 pendingRoutingConfig.RampingDeploymentVersion = nil
1398
1399 // TODO (Shivam): remove the empty string check once canary stops flaking out
1400 > if newCurrentVersion != worker_versioning.UnversionedVersionId && newCurrentVersion != "" { workflow.go
1401 // Tell new current version that it's current
1402 currUpdateArgs := &deploymentspb.SyncVersionStateUpdateArgs{
go.temporal.io/server/api/deployment/v1/message.pb.go 14 introduced LOC · 2 ranges

Open complete file

2271 }
2272
2273 > func (x *SetCurrentVersionArgs) Reset() { message.pb.go
2274 > *x = SetCurrentVersionArgs{}
2275 > mi := &file_temporal_server_api_deployment_v1_message_proto_msgTypes[32]
2276 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
2277 > ms.StoreMessageInfo(mi)
2278 > }
2279
2280 func (x *SetCurrentVersionArgs) String() string {
2284 func (*SetCurrentVersionArgs) ProtoMessage() {}
2285
2286 > func (x *SetCurrentVersionArgs) ProtoReflect() protoreflect.Message { message.pb.go
2287 > mi := &file_temporal_server_api_deployment_v1_message_proto_msgTypes[32]
2288 > if x != nil {
2289 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
2290 > if ms.LoadMessageInfo() == nil {
2291 > ms.StoreMessageInfo(mi)
2292 > }
2293 > return ms
2294 }
2295 return mi.MessageOf(x)

Direct hierarchy

Parents

Children