Merge branch 'main' into ss/rate_limiter_tqptm

Author: Shivs11

service/worker/migration/activities.go

@@ -424,6 +424,7 @@ func (a *activities) UpdateActiveCluster(ctx context.Context, req updateActiveCl
 
 func (a *activities) ListWorkflows(ctx context.Context, request *workflowservice.ListWorkflowExecutionsRequest) (*listWorkflowsResponse, error) {
 	ctx = headers.SetCallerInfo(ctx, headers.NewCallerInfo(request.Namespace, headers.CallerTypePreemptable, ""))
+	ctx = metadata.NewOutgoingContext(ctx, metadata.Pairs(interceptor.DCRedirectionContextHeaderName, "false"))
 
 	resp, err := a.frontendClient.ListWorkflowExecutions(ctx, request)
 	if err != nil {

service/worker/replicator/namespace_replication_message_processor.go renamed to service/worker/replicator/replication_message_processor.go

@@ -33,7 +33,7 @@ const (
 	taskProcessorErrorRetryMaxAttampts        = 5
 )
 
-func newNamespaceReplicationMessageProcessor(
+func newReplicationMessageProcessor(
 	currentCluster string,
 	sourceCluster string,
 	logger log.Logger,
@@ -45,12 +45,12 @@ func newNamespaceReplicationMessageProcessor(
 	namespaceReplicationQueue persistence.NamespaceReplicationQueue,
 	matchingClient matchingservice.MatchingServiceClient,
 	namespaceRegistry namespace.Registry,
-) *namespaceReplicationMessageProcessor {
+) *replicationMessageProcessor {
 	retryPolicy := backoff.NewExponentialRetryPolicy(taskProcessorErrorRetryWait).
 		WithBackoffCoefficient(taskProcessorErrorRetryBackoffCoefficient).
 		WithMaximumAttempts(taskProcessorErrorRetryMaxAttampts)
 
-	return &namespaceReplicationMessageProcessor{
+	return &replicationMessageProcessor{
 		hostInfo:                  hostInfo,
 		serviceResolver:           serviceResolver,
 		status:                    common.DaemonStatusInitialized,
@@ -71,7 +71,7 @@ func newNamespaceReplicationMessageProcessor(
 }
 
 type (
-	namespaceReplicationMessageProcessor struct {
+	replicationMessageProcessor struct {
 		hostInfo                  membership.HostInfo
 		serviceResolver           membership.ServiceResolver
 		status                    int32
@@ -91,21 +91,21 @@ type (
 	}
 )
 
-func (p *namespaceReplicationMessageProcessor) Start() {
+func (p *replicationMessageProcessor) Start() {
 	if !atomic.CompareAndSwapInt32(&p.status, common.DaemonStatusInitialized, common.DaemonStatusStarted) {
 		return
 	}
 
 	go p.processorLoop()
 }
 
-func (p *namespaceReplicationMessageProcessor) processorLoop() {
+func (p *replicationMessageProcessor) processorLoop() {
 	timer := time.NewTimer(getWaitDuration())
 
 	for {
 		select {
 		case <-timer.C:
-			p.getAndHandleNamespaceReplicationTasks()
+			p.handleReplicationTasks()
 			timer.Reset(getWaitDuration())
 		case <-p.done:
 			timer.Stop()
@@ -114,7 +114,7 @@ func (p *namespaceReplicationMessageProcessor) processorLoop() {
 	}
 }
 
-func (p *namespaceReplicationMessageProcessor) getAndHandleNamespaceReplicationTasks() {
+func (p *replicationMessageProcessor) handleReplicationTasks() {
 	// The following is a best effort to make sure only one worker is processing tasks for a
 	// particular source cluster. When the ring is under reconfiguration, it is possible that
 	// for a small period of time two or more workers think they are the owner and try to execute
@@ -153,12 +153,12 @@ func (p *namespaceReplicationMessageProcessor) getAndHandleNamespaceReplicationT
 	for taskIndex := range response.Messages.ReplicationTasks {
 		task := response.Messages.ReplicationTasks[taskIndex]
 		err := backoff.ThrottleRetry(func() error {
-			return p.handleNamespaceReplicationTask(taskCtx, task)
+			return p.handleReplicationTask(taskCtx, task)
 		}, p.retryPolicy, isTransientRetryableError)
 
 		if err != nil {
 			metrics.ReplicatorFailures.With(p.metricsHandler).Record(1)
-			p.logger.Error("Failed to apply namespace replication tasks", tag.Error(err))
+			p.logger.Error("Failed to apply replication tasks", tag.Error(err))
 
 			dlqErr := backoff.ThrottleRetry(func() error {
 
@@ -176,7 +176,7 @@ func (p *namespaceReplicationMessageProcessor) getAndHandleNamespaceReplicationT
 	p.lastRetrievedMessageID = response.Messages.GetLastRetrievedMessageId()
 }
 
-func (p *namespaceReplicationMessageProcessor) putNamespaceReplicationTaskToDLQ(
+func (p *replicationMessageProcessor) putNamespaceReplicationTaskToDLQ(
 	ctx context.Context,
 	task *replicationspb.ReplicationTask,
 ) error {
@@ -205,7 +205,7 @@ func (p *namespaceReplicationMessageProcessor) putNamespaceReplicationTaskToDLQ(
 	return p.namespaceReplicationQueue.PublishToDLQ(ctx, task)
 }
 
-func (p *namespaceReplicationMessageProcessor) handleNamespaceReplicationTask(
+func (p *replicationMessageProcessor) handleReplicationTask(
 	ctx context.Context,
 	task *replicationspb.ReplicationTask,
 ) error {
@@ -222,23 +222,25 @@ func (p *namespaceReplicationMessageProcessor) handleNamespaceReplicationTask(
 		err := p.namespaceTaskExecutor.Execute(ctx, attr)
 		if err != nil {
 			p.logger.Error("unable to process namespace replication task",
-				tag.WorkflowNamespaceID(attr.Id))
+				tag.WorkflowNamespaceID(attr.Id),
+				tag.Error(err))
 		}
 		return err
 	case enumsspb.REPLICATION_TASK_TYPE_TASK_QUEUE_USER_DATA:
 		attr := task.GetTaskQueueUserDataAttributes()
 		err := p.handleTaskQueueUserDataReplicationTask(ctx, attr)
 		if err != nil {
 			p.logger.Error(fmt.Sprintf("unable to process task queue metadata replication task, %v", attr.TaskQueueName),
-				tag.WorkflowNamespaceID(attr.NamespaceId))
+				tag.WorkflowNamespaceID(attr.NamespaceId),
+				tag.Error(err))
 		}
 		return err
 	default:
 		return fmt.Errorf("cannot handle replication task of type %v", task.TaskType)
 	}
 }
 
-func (p *namespaceReplicationMessageProcessor) handleTaskQueueUserDataReplicationTask(
+func (p *replicationMessageProcessor) handleTaskQueueUserDataReplicationTask(
 	ctx context.Context,
 	attrs *replicationspb.TaskQueueUserDataAttributes,
 ) error {
@@ -266,7 +268,7 @@ func (p *namespaceReplicationMessageProcessor) handleTaskQueueUserDataReplicatio
 	return err
 }
 
-func (p *namespaceReplicationMessageProcessor) Stop() {
+func (p *replicationMessageProcessor) Stop() {
 	close(p.done)
 }
 

service/worker/replicator/replicator.go

@@ -40,7 +40,7 @@ type (
 		replicationCleanupGroup          goro.Group
 
 		namespaceProcessorsLock sync.Mutex
-		namespaceProcessors     map[string]*namespaceReplicationMessageProcessor
+		namespaceProcessors     map[string]*replicationMessageProcessor
 		matchingClient          matchingservice.MatchingServiceClient
 		namespaceRegistry       namespace.Registry
 	}
@@ -69,7 +69,7 @@ func NewReplicator(
 		serviceResolver:                  serviceResolver,
 		clusterMetadata:                  clusterMetadata,
 		namespaceReplicationTaskExecutor: namespaceReplicationTaskExecutor,
-		namespaceProcessors:              make(map[string]*namespaceReplicationMessageProcessor),
+		namespaceProcessors:              make(map[string]*replicationMessageProcessor),
 		clientBean:                       clientBean,
 		logger:                           log.With(logger, tag.ComponentReplicator),
 		metricsHandler:                   metricsHandler,
@@ -139,7 +139,7 @@ func (r *Replicator) listenToClusterMetadataChange() {
 						// This should never happen as cluster metadata should have the up-to-date data.
 						panic(fmt.Sprintf("Bug found in cluster metadata with error %v", err))
 					}
-					processor := newNamespaceReplicationMessageProcessor(
+					processor := newReplicationMessageProcessor(
 						currentClusterName,
 						clusterName,
 						log.With(r.logger, tag.ComponentReplicationTaskProcessor, tag.SourceCluster(clusterName)),

service/worker/scheduler/spec.go

@@ -276,7 +276,7 @@ func (cs *CompiledSpec) GetNextTime(jitterSeed string, after time.Time) GetNextT
 	after = util.MaxTime(after, cs.spec.StartTime.AsTime().Add(-time.Second))
 
 	pastEndTime := func(t time.Time) bool {
-		return cs.spec.EndTime != nil && t.After(cs.spec.EndTime.AsTime())
+		return cs.spec.EndTime != nil && t.After(cs.spec.EndTime.AsTime()) || t.Year() > maxCalendarYear
 	}
 	var nominal time.Time
 	for nominal.IsZero() || cs.excluded(nominal) {

service/worker/scheduler/spec_test.go

@@ -366,6 +366,19 @@ func (s *specSuite) TestSpecExclude() {
 	)
 }
 
+func (s *specSuite) TestExcludeAll() {
+	cs, err := s.specBuilder.NewCompiledSpec(&schedulepb.ScheduleSpec{
+		Interval: []*schedulepb.IntervalSpec{
+			{Interval: durationpb.New(7 * 24 * time.Hour)},
+		},
+		ExcludeCalendar: []*schedulepb.CalendarSpec{
+			&schedulepb.CalendarSpec{Second: "*", Minute: "*", Hour: "*"},
+		},
+	})
+	s.NoError(err)
+	s.Zero(cs.GetNextTime("", time.Date(2022, 3, 23, 12, 53, 2, 9, time.UTC)))
+}
+
 func (s *specSuite) TestSpecStartTime() {
 	s.checkSequenceFull(
 		"",

service/worker/workerdeployment/workflow.go

@@ -127,13 +127,30 @@ func (d *WorkflowRunner) updateVersionSummary(summary *deploymentspb.WorkerDeplo
 }
 
 func (d *WorkflowRunner) run(ctx workflow.Context) error {
-	if d.GetState().GetCreateTime() == nil {
+	// TODO(carlydf): remove verbose logging
+	d.logger.Info("Raw workflow state at start",
+		"state_nil", d.State == nil,
+		"create_time_nil", d.GetState().GetCreateTime() == nil,
+		"routing_config_nil", d.GetState().GetRoutingConfig() == nil,
+		"raw_state", d.State,
+		"workflow_id", workflow.GetInfo(ctx).WorkflowExecution.ID,
+		"run_id", workflow.GetInfo(ctx).WorkflowExecution.RunID)
+
+	if d.GetState().GetCreateTime() == nil ||
+		d.GetState().GetRoutingConfig() == nil ||
+		d.GetState().GetConflictToken() == nil {
 		if d.State == nil {
 			d.State = &deploymentspb.WorkerDeploymentLocalState{}
 		}
-		d.State.CreateTime = timestamppb.New(workflow.Now(ctx))
-		d.State.RoutingConfig = &deploymentpb.RoutingConfig{CurrentVersion: worker_versioning.UnversionedVersionId}
-		d.State.ConflictToken, _ = workflow.Now(ctx).MarshalBinary()
+		if d.State.CreateTime == nil {
+			d.State.CreateTime = timestamppb.New(workflow.Now(ctx))
+		}
+		if d.State.RoutingConfig == nil {
+			d.State.RoutingConfig = &deploymentpb.RoutingConfig{CurrentVersion: worker_versioning.UnversionedVersionId}
+		}
+		if d.State.ConflictToken == nil {
+			d.State.ConflictToken, _ = workflow.Now(ctx).MarshalBinary()
+		}
 
 		// updating the memo since the RoutingConfig is updated
 		if err := d.updateMemo(ctx); err != nil {
@@ -144,6 +161,15 @@ func (d *WorkflowRunner) run(ctx workflow.Context) error {
 		d.State.Versions = make(map[string]*deploymentspb.WorkerDeploymentVersionSummary)
 	}
 
+	// TODO(carlydf): remove verbose logging
+	d.logger.Info("Starting workflow run",
+		"create_time", d.State.GetCreateTime(),
+		"routing_config", d.State.GetRoutingConfig(),
+		//nolint:staticcheck // SA1019: worker versioning v0.31
+		"current_version", d.State.GetRoutingConfig().GetCurrentVersion(),
+		//nolint:staticcheck // SA1019: worker versioning v0.31
+		"ramping_version", d.State.GetRoutingConfig().GetRampingVersion())
+
 	err := workflow.SetQueryHandler(ctx, QueryDescribeDeployment, func() (*deploymentspb.QueryDescribeWorkerDeploymentResponse, error) {
 		return &deploymentspb.QueryDescribeWorkerDeploymentResponse{
 			State: d.State,
@@ -224,10 +250,25 @@ func (d *WorkflowRunner) run(ctx workflow.Context) error {
 	// Wait until we can continue as new or are cancelled. The workflow will continue-as-new iff
 	// there are no pending updates/signals and the state has changed.
 	err = workflow.Await(ctx, func() bool {
-		return d.deleteDeployment || // deployment is deleted -> it's ok to drop all signals and updates.
+		canContinue := d.deleteDeployment || // deployment is deleted -> it's ok to drop all signals and updates.
 			// There is no pending signal or update, but the state is dirty or forceCaN is requested:
 			(!d.signalHandler.signalSelector.HasPending() && d.signalHandler.processingSignals == 0 && workflow.AllHandlersFinished(ctx) &&
 				(d.forceCAN || d.stateChanged))
+
+		// TODO(carlydf): remove verbose logging
+		if canContinue {
+			d.logger.Info("Workflow can continue as new",
+				"workflow_id", workflow.GetInfo(ctx).WorkflowExecution.ID,
+				"run_id", workflow.GetInfo(ctx).WorkflowExecution.RunID,
+				"delete_deployment", d.deleteDeployment,
+				"has_pending_signals", d.signalHandler.signalSelector.HasPending(),
+				"processing_signals", d.signalHandler.processingSignals,
+				"all_handlers_finished", workflow.AllHandlersFinished(ctx),
+				"force_can", d.forceCAN,
+				"state_changed", d.stateChanged,
+				"routing_config", d.State.GetRoutingConfig())
+		}
+		return canContinue
 	})
 	if err != nil {
 		return err
@@ -237,6 +278,19 @@ func (d *WorkflowRunner) run(ctx workflow.Context) error {
 		return nil
 	}
 
+	// TODO(carlydf): remove verbose logging
+	d.logger.Info("Continuing workflow as new",
+		"create_time", d.State.GetCreateTime(),
+		"routing_config", d.State.GetRoutingConfig(),
+		//nolint:staticcheck // SA1019: worker versioning v0.31
+		"current_version", d.State.GetRoutingConfig().GetCurrentVersion(),
+		//nolint:staticcheck // SA1019: worker versioning v0.31
+		"ramping_version", d.State.GetRoutingConfig().GetRampingVersion(),
+		"state_changed", d.stateChanged,
+		"force_can", d.forceCAN,
+		"workflow_id", workflow.GetInfo(ctx).WorkflowExecution.ID,
+		"run_id", workflow.GetInfo(ctx).WorkflowExecution.RunID)
+
 	// We perform a continue-as-new after each update and signal is handled to ensure compatibility
 	// even if the server rolls back to a previous minor version. By continuing-as-new,
 	// we pass the current state as input to the next workflow execution, resulting in a new
@@ -338,19 +392,23 @@ func (d *WorkflowRunner) handleDeleteDeployment(ctx workflow.Context) error {
 }
 
 func (d *WorkflowRunner) validateStateBeforeAcceptingRampingUpdate(args *deploymentspb.SetRampingVersionArgs) error {
-	if args.Version == d.State.RoutingConfig.RampingVersion && args.Percentage == d.State.RoutingConfig.RampingVersionPercentage && args.Identity == d.State.LastModifierIdentity {
-		return temporal.NewApplicationError("version already ramping, no change", errNoChangeType, d.State.ConflictToken)
+	//nolint:staticcheck // SA1019: worker versioning v0.31
+	if args.Version == d.State.GetRoutingConfig().GetRampingVersion() &&
+		args.Percentage == d.State.GetRoutingConfig().GetRampingVersionPercentage() &&
+		args.Identity == d.State.GetLastModifierIdentity() {
+		return temporal.NewApplicationError("version already ramping, no change", errNoChangeType, d.State.GetConflictToken())
 	}
 
-	if args.ConflictToken != nil && !bytes.Equal(args.ConflictToken, d.State.ConflictToken) {
+	if args.ConflictToken != nil && !bytes.Equal(args.ConflictToken, d.State.GetConflictToken()) {
 		return temporal.NewApplicationError("conflict token mismatch", errFailedPrecondition)
 	}
-	if args.Version == d.State.RoutingConfig.CurrentVersion {
+	//nolint:staticcheck // SA1019: worker versioning v0.31
+	if args.Version == d.State.GetRoutingConfig().GetCurrentVersion() {
 		d.logger.Info("version can't be set to ramping since it is already current")
 		return temporal.NewApplicationError(fmt.Sprintf("requested ramping version %s is already current", args.Version), errFailedPrecondition)
 	}
 
-	if _, ok := d.State.Versions[args.Version]; !ok && args.Version != "" && args.Version != worker_versioning.UnversionedVersionId {
+	if _, ok := d.State.GetVersions()[args.Version]; !ok && args.Version != "" && args.Version != worker_versioning.UnversionedVersionId {
 		d.logger.Info("version not found in deployment")
 		return temporal.NewApplicationError(fmt.Sprintf("requested ramping version %s not found in deployment", args.Version), errFailedPrecondition)
 	}
@@ -579,7 +637,8 @@ func (d *WorkflowRunner) handleDeleteVersion(ctx workflow.Context, args *deploym
 }
 
 func (d *WorkflowRunner) validateStateBeforeAcceptingSetCurrent(args *deploymentspb.SetCurrentVersionArgs) error {
-	if d.State.RoutingConfig.CurrentVersion == args.Version && d.State.LastModifierIdentity == args.Identity {
+	//nolint:staticcheck // SA1019: worker versioning v0.31
+	if d.State.GetRoutingConfig().GetCurrentVersion() == args.Version && d.State.GetLastModifierIdentity() == args.Identity {
 		return temporal.NewApplicationError("no change", errNoChangeType, d.State.ConflictToken)
 	}
 	if args.ConflictToken != nil && !bytes.Equal(args.ConflictToken, d.State.ConflictToken) {
@@ -609,6 +668,14 @@ func (d *WorkflowRunner) handleSetCurrent(ctx workflow.Context, args *deployment
 		d.lock.Unlock()
 	}()
 
+	// Log state before update
+	// TODO(carlydf): remove verbose logging
+	d.logger.Info("Starting SetCurrent update",
+		//nolint:staticcheck // SA1019: worker versioning v0.31
+		"current_version", d.State.GetRoutingConfig().GetCurrentVersion(),
+		"new_version", args.Version,
+		"routing_config", d.State.GetRoutingConfig())
+
 	// Validating the state before starting the SetCurrent operation. This is required due to the following reason:
 	// The validator accepts/rejects updates based on the state of the deployment workflow. Theoretically, two concurrent update requests
 	// 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
@@ -955,6 +1022,14 @@ func (d *WorkflowRunner) newUUID(ctx workflow.Context) string {
 }
 
 func (d *WorkflowRunner) updateMemo(ctx workflow.Context) error {
+	// TODO(carlydf): remove verbose logging
+	d.logger.Info("Updating workflow memo",
+		"routing_config", d.State.GetRoutingConfig(),
+		//nolint:staticcheck // SA1019: worker versioning v0.31
+		"current_version", d.State.GetRoutingConfig().GetCurrentVersion(),
+		//nolint:staticcheck // SA1019: worker versioning v0.31
+		"ramping_version", d.State.GetRoutingConfig().GetRampingVersion())
+
 	return workflow.UpsertMemo(ctx, map[string]any{
 		WorkerDeploymentMemoField: &deploymentspb.WorkerDeploymentWorkflowMemo{
 			DeploymentName: d.DeploymentName,