tasks/service/history/api/pauseworkflow/api.go

package pauseworkflow

import (
	"context"

	enumspb "go.temporal.io/api/enums/v1"
	"go.temporal.io/api/serviceerror"
	"go.temporal.io/api/workflowservice/v1"
	"go.temporal.io/server/api/historyservice/v1"
	"go.temporal.io/server/common/definition"
	"go.temporal.io/server/common/namespace"
	"go.temporal.io/server/service/history/api"
	"go.temporal.io/server/service/history/consts"
	historyi "go.temporal.io/server/service/history/interfaces"
)

func Invoke(
	ctx context.Context,
	req *historyservice.PauseWorkflowExecutionRequest,
	shard historyi.ShardContext,
	workflowConsistencyChecker api.WorkflowConsistencyChecker,
) (resp *historyservice.PauseWorkflowExecutionResponse, retError error) {
	namespaceEntry, err := api.GetActiveNamespace(shard, namespace.ID(req.GetNamespaceId()), req.GetPauseRequest().GetWorkflowId())
	if err != nil {
		return nil, err
	}
	namespaceID := namespaceEntry.ID()

	pauseRequest := req.GetPauseRequest()
	if pauseRequest == nil {
		return nil, serviceerror.NewInvalidArgument("pause request is nil.")
	}

	if err := validatePauseRequest(pauseRequest, shard); err != nil {
		return nil, err
	}

	err = api.GetAndUpdateWorkflowWithNew(
		ctx,
		nil,
		definition.NewWorkflowKey(
			namespaceID.String(),
			pauseRequest.GetWorkflowId(),
			pauseRequest.GetRunId(),
		),
		func(workflowLease api.WorkflowLease) (*api.UpdateWorkflowAction, error) {
			mutableState := workflowLease.GetMutableState()

			releaseFn := workflowLease.GetReleaseFn()
			// Make sure the workflow is not closed.
			if !mutableState.IsWorkflowExecutionRunning() {
				// in-memory mutable state is still clean, release the lock with nil error to prevent
				// clearing and reloading mutable state
				releaseFn(nil)
				return nil, consts.ErrWorkflowCompleted
			}

			// Check if workflow is already paused
			if mutableState.GetExecutionState().GetStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_PAUSED {
				pauseInfo := mutableState.GetExecutionInfo().PauseInfo
				if pauseInfo != nil && pauseInfo.RequestId != "" && pauseInfo.RequestId == pauseRequest.GetRequestId() {
					// Already paused with the same request id, nothing to do
					return &api.UpdateWorkflowAction{
						Noop:               true,
						CreateWorkflowTask: false,
					}, nil
				}
				releaseFn(nil)
				return nil, serviceerror.NewFailedPrecondition("workflow is already paused.")
			}

			// Add the workflow execution paused event
			_, err := mutableState.AddWorkflowExecutionPausedEvent(
				pauseRequest.GetIdentity(),
				pauseRequest.GetReason(),
				pauseRequest.GetRequestId(),
			)
			if err != nil {
				return nil, err
			}

			return &api.UpdateWorkflowAction{
				Noop:               false,
				CreateWorkflowTask: false, // Don't create workflow task when pausing
			}, nil
		},
		nil,
		shard,
		workflowConsistencyChecker,
	)
	if err != nil {
		return nil, err
	}

	// Return response
	return &historyservice.PauseWorkflowExecutionResponse{}, nil
}

func validatePauseRequest(req *workflowservice.PauseWorkflowExecutionRequest, shard historyi.ShardContext) error {
	// verify size limits of reason, request id and identity.
	if len(req.GetReason()) > shard.GetConfig().MaxIDLengthLimit() {
		return serviceerror.NewInvalidArgument("reason is too long.")
	}
	if len(req.GetRequestId()) > shard.GetConfig().MaxIDLengthLimit() {
		return serviceerror.NewInvalidArgument("request id is too long.")
	}
	if len(req.GetIdentity()) > shard.GetConfig().MaxIDLengthLimit() {
		return serviceerror.NewInvalidArgument("identity is too long.")
	}

	return nil
}