version_workflow.go ×22

Frontier kind: Code frontier

unlabeled · c_30b830a4894c

290 tests · 2712 LOC · 133 files · introduces 0 tests · 122 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code: 27 ranges122 lines · 2 files
Tests: 0 tests

Contains — complete concept membership

All code (extent): 405 ranges2712 lines · 133 files · Browse complete extent
All tests (intent): 290 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.

Focus concept
Ancestors (broader)
Descendants (narrower)
Introduced files
Introduced tests

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: 122 introduced LOC across 27 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/worker/workerdeployment/version_workflow.go 96 introduced LOC · 22 ranges

Open complete file

  80
  
  	unsafeVisibilityGracePeriodGetter func() time.Duration,
  
  81
  
  	versionWorkflowArgs *deploymentspb.WorkerDeploymentVersionWorkflowArgs,
  
  82
  >
  ) error {
  
version_workflow.go

  83
  >
  	versionWorkflowRunner := &VersionWorkflowRunner{
  
  84
  >
  		WorkerDeploymentVersionWorkflowArgs: versionWorkflowArgs,
  
  85
  >
  
  86
  >
  		a:                                 nil,
  
  87
  >
  		logger:                            sdklog.With(workflow.GetLogger(ctx), "wf-namespace", versionWorkflowArgs.NamespaceName),
  
  88
  >
  		metrics:                           workflow.GetMetricsHandler(ctx).WithTags(map[string]string{"namespace": versionWorkflowArgs.NamespaceName}),
  
  89
  >
  		lock:                              workflow.NewMutex(ctx),
  
  90
  >
  		computeConfigLock:                 workflow.NewMutex(ctx),
  
  91
  >
  		unsafeWorkflowVersionGetter:       unsafeWorkflowVersionGetter,
  
  92
  >
  		unsafeRefreshIntervalGetter:       unsafeRefreshIntervalGetter,
  
  93
  >
  		unsafeVisibilityGracePeriodGetter: unsafeVisibilityGracePeriodGetter,
  
  94
  >
  		signalHandler: &SignalHandler{
  
  95
  >
  			signalSelector: workflow.NewSelector(ctx),
  
  96
  >
  		},
  
  97
  >
  		workflowVersion: getWorkflowVersion(ctx, unsafeWorkflowVersionGetter),
  
  98
  >
  	}
  
  99
  >
  	return versionWorkflowRunner.run(ctx)
  
  100
  >
  }
  
  101
  
  102
  >
  func (d *VersionWorkflowRunner) listenToSignals(ctx workflow.Context) {
  
version_workflow.go

  103
  >
  	// Fetch signal channels
  
  104
  >
  	forceCANSignalChannel := workflow.GetSignalChannel(ctx, ForceCANSignalName)
  
  105
  >
  	drainageStatusSignalChannel := workflow.GetSignalChannel(ctx, SyncDrainageSignalName)
  
  106
  >
  
  107
  >
  	d.signalHandler.signalSelector.AddReceive(forceCANSignalChannel, func(c workflow.ReceiveChannel, more bool) {
  
  108
  
  		d.signalHandler.processingSignals++
  
  109
  
  		defer func() { d.signalHandler.processingSignals-- }()

  118
  
  		}
  
  119
  
  	})
  
  120
  >
  	d.signalHandler.signalSelector.AddReceive(drainageStatusSignalChannel, func(c workflow.ReceiveChannel, more bool) {
  
version_workflow.go

  121
  
  		d.signalHandler.processingSignals++
  
  122
  
  		defer func() { d.signalHandler.processingSignals-- }()

  149
  
  150
  
  	// Version gate for reactivation signal to prevent NDEs during rollback
  
  151
  >
  	if workflow.GetVersion(ctx, "reactivation-signal", workflow.DefaultVersion, 0) >= 0 {
  
version_workflow.go

  152
  
  		reactivateSignalChannel := workflow.GetSignalChannel(ctx, ReactivateVersionSignalName)
  
  153

  186
  
  187
  
  	// Version gate for demote version signal to prevent NDEs during rollback
  
  188
  >
  	if workflow.GetVersion(ctx, "demote-version-signal", workflow.DefaultVersion, 0) >= 0 {
  
version_workflow.go

  189
  
  		demoteSignalChannel := workflow.GetSignalChannel(ctx, DemoteVersionSignalName)
  
  190

  230
  
  231
  
  	// Version gate for sync-validation-status signal to prevent NDEs during rollback
  
  232
  >
  	if workflow.GetVersion(ctx, "sync-validation-status-signal", workflow.DefaultVersion, 0) >= 0 {
  
version_workflow.go

  233
  
  		syncValidationStatusChannel := workflow.GetSignalChannel(ctx, worker_versioning.SignalSyncValidationStatus)
  
  234
  
  		d.signalHandler.signalSelector.AddReceive(syncValidationStatusChannel, func(c workflow.ReceiveChannel, more bool) {

  244
  
  245
  
  	// Keep waiting for signals, when it's time to CaN the main goroutine will exit.
  
  246
  >
  	for {
  
version_workflow.go

  247
  >
  		d.signalHandler.signalSelector.Select(ctx)
  
  248
  >
  	}
  
  249
  
  }
  
  250
  
  251
  >
  func (d *VersionWorkflowRunner) run(ctx workflow.Context) error {
  
version_workflow.go

  252
  >
  	if d.GetVersionState().Version == nil {
  
  253
  
  		return fmt.Errorf("version cannot be nil on start")
  
  254
  
  	}
  
  255
  >
  	if d.VersionState.GetCreateTime() == nil {
  
version_workflow.go

  256
  
  		d.VersionState.CreateTime = timestamppb.New(workflow.Now(ctx))
  
  257
  
  	}
  
  258
  
  	// TODO: remove this after next release because now the status should always be set at start.
  
  259
  >
  	if d.VersionState.Status == enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_UNSPECIFIED {
  
version_workflow.go

  260
  
  		d.VersionState.Status = enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE
  
  261
  
  	}
  
  262
  
  263
  
  	// if we were draining and just continued-as-new, do another drainage check after waiting for appropriate time
  
  264
  >
  	if d.VersionState.GetDrainageInfo().GetStatus() == enumspb.VERSION_DRAINAGE_STATUS_DRAINING {
  
version_workflow.go

  265
  
  		workflow.Go(ctx, d.refreshDrainageInfo)
  
  266
  
  	}
  
  267
  
  268
  
  	// Set up Query Handlers here:
  
  269
  >
  	if err := workflow.SetQueryHandler(ctx, QueryDescribeVersion, d.handleDescribeQuery); err != nil {
  
version_workflow.go

  270
  
  		d.logger.Error("Failed while setting up query handler")
  
  271
  
  		return err
  
  272
  
  	}
  
  273
  
  274
  >
  	if err := workflow.SetUpdateHandlerWithOptions(
  
version_workflow.go

  275
  >
  		ctx,
  
  276
  >
  		RegisterWorkerInDeploymentVersion,
  
  277
  >
  		d.handleRegisterWorker,
  
  278
  >
  		workflow.UpdateHandlerOptions{
  
  279
  >
  			Validator: d.validateRegisterWorker,
  
  280
  >
  		},
  
  281
  >
  	); err != nil {
  
  282
  
  		return err
  
  283
  
  	}
  
  284
  
  285
  >
  	if err := workflow.SetUpdateHandlerWithOptions(
  
version_workflow.go

  286
  >
  		ctx,
  
  287
  >
  		SyncVersionState,
  
  288
  >
  		d.handleSyncState,
  
  289
  >
  		workflow.UpdateHandlerOptions{
  
  290
  >
  			Validator: d.validateSyncState,
  
  291
  >
  		},
  
  292
  >
  	); err != nil {
  
  293
  
  		return err
  
  294
  
  	}
  
  295
  
  296
  >
  	if err := workflow.SetUpdateHandlerWithOptions(
  
version_workflow.go

  297
  >
  		ctx,
  
  298
  >
  		DeleteVersion,
  
  299
  >
  		d.handleDeleteVersion,
  
  300
  >
  		workflow.UpdateHandlerOptions{
  
  301
  >
  			Validator: d.validateDeleteVersion,
  
  302
  >
  		},
  
  303
  >
  	); err != nil {
  
  304
  
  		return err
  
  305
  
  	}
  
  306
  
  307
  >
  	if err := workflow.SetUpdateHandlerWithOptions(
  
version_workflow.go

  308
  >
  		ctx,
  
  309
  >
  		UpdateVersionMetadata,
  
  310
  >
  		d.handleUpdateVersionMetadata,
  
  311
  >
  		workflow.UpdateHandlerOptions{
  
  312
  >
  			Validator: d.validateUpdateVersionMetadata,
  
  313
  >
  		},
  
  314
  >
  	); err != nil {
  
  315
  
  		return err
  
  316
  
  	}
  
  317
  
  318
  >
  	if err := workflow.SetUpdateHandlerWithOptions(
  
version_workflow.go

  319
  >
  		ctx,
  
  320
  >
  		UpdateVersionComputeConfig,
  
  321
  >
  		d.handleUpdateVersionComputeConfig,
  
  322
  >
  		workflow.UpdateHandlerOptions{
  
  323
  >
  			Validator: d.validateUpdateVersionComputeConfig,
  
  324
  >
  		},
  
  325
  >
  	); err != nil {
  
  326
  
  		return err
  
  327
  
  	}

  329
  
  	// Deployment workflow should always be running before starting the version workflow.
  
  330
  
  	// We should not start the deployment workflow. If we cannot find the deployment workflow when signaling, it means a bug and we should fix it.
  
  331
  >
  	if !d.hasMinVersion(VersionDataRevisionNumber) {
  
version_workflow.go

  332
  
  		// First ensure deployment workflow is running
  
  333
  
  		//nolint:staticcheck // SA1019

  348
  
  	// When creating a compute provider and version together, there is a race condition between the two coming up. Making sure to have pulled
  
  349
  
  	// the latest state from the compute provider if this happens to be the slower one.
  
  350
  >
  	if err := d.syncVersionDataToComputeStatus(ctx); err != nil {
  
version_workflow.go

  351
  
  		return err
  
  352
  
  	}
  
  353
  
  354
  
  	// Listen to signals in a different goroutine to make business logic clearer
  
  355
  >
  	workflow.Go(ctx, d.listenToSignals)
  
version_workflow.go

  356
  >
  
  357
  >
  	// Wait until we can continue as new or are cancelled. The workflow will continue-as-new iff
  
  358
  >
  	// there are no pending updates/signals and the state has changed.
  
  359
  >
  	err := workflow.Await(ctx, func() bool {
  
  360
  >
  		return (d.deleteVersion && d.asyncPropagationsInProgress == 0) || // version is deleted -> it's ok to drop all signals and updates.
  
  361
  >
  			// There is no pending signal or update, but the state is dirty or forceCaN is requested:
  
  362
  >
  			(!d.signalHandler.signalSelector.HasPending() && d.signalHandler.processingSignals == 0 && workflow.AllHandlersFinished(ctx) &&
  
  363
  >
  				// And there is a force CaN or a propagated state change or history got too large
  
  364
  >
  				(d.forceCAN || (d.stateChanged && d.asyncPropagationsInProgress == 0) || workflow.GetInfo(ctx).GetContinueAsNewSuggested()))
  
  365
  >
  	})
  
  366
  
  	if err != nil {
  
  367
  
  		return err

  1252
  
  1253
  
  // syncVersionDataToComputeStatus is a helper that syncs the compute status from WCI to the worker deployment version
  
  1254
  >
  func (d *VersionWorkflowRunner) syncVersionDataToComputeStatus(ctx workflow.Context) error {
  
version_workflow.go

  1255
  >
  	if workflow.GetVersion(ctx, "sync-compute-validation-status", workflow.DefaultVersion, 0) == workflow.DefaultVersion {
  
  1256
  
  		return nil
  
  1257
  
  	}

  1515
  
  }
  
  1516
  
  1517
  >
  func (d *VersionWorkflowRunner) hasMinVersion(version DeploymentWorkflowVersion) bool {
  
version_workflow.go

  1518
  >
  	return d.workflowVersion >= version
  
  1519
  >
  }
  
  1520
  
  1521
  
  // syncRegisteredTaskQueueAsync syncs the routing config and version data to the new task queue.

go.temporal.io/server/api/deployment/v1/message.pb.go 26 introduced LOC · 5 ranges