func (wh *WorkflowHandler) RespondActivityTaskCanceled(
	ctx context.Context,
	cancelRequest *types.RespondActivityTaskCanceledRequest,
) (retError error) {
	defer log.CapturePanic(wh.GetLogger(), &retError)

	scope := wh.getDefaultScope(ctx, metrics.FrontendRespondActivityTaskCanceledScope)

	if wh.isShuttingDown() {
		return errShuttingDown
	}

	if err := wh.versionChecker.ClientSupported(ctx, wh.config.EnableClientVersionCheck()); err != nil {
		return wh.error(err, scope)
	}

	if cancelRequest == nil {
		return wh.error(errRequestNotSet, scope)
	}

	if cancelRequest.TaskToken == nil {
		return wh.error(errTaskTokenNotSet, scope)
	}

	taskToken, err := wh.tokenSerializer.Deserialize(cancelRequest.TaskToken)
	if err != nil {
		return wh.error(err, scope)
	}

	if taskToken.DomainID == "" {
		return wh.error(errDomainNotSet, scope)
	}

	domainName, err := wh.GetDomainCache().GetDomainName(taskToken.DomainID)
	if err != nil {
		return wh.error(err, scope)
	}

	dw := domainWrapper{
		domain: domainName,
	}
	scope, sw := wh.startRequestProfileWithDomain(
		ctx,
		metrics.FrontendRespondActivityTaskCanceledScope,
		dw,
	)
	defer sw.Stop()

	// Count the request in the host RPS,
	// but we still accept it even if RPS is exceeded
	wh.allow(false, dw)

	tags := getDomainWfIDRunIDTags(domainName, &types.WorkflowExecution{
		WorkflowID: taskToken.WorkflowID,
		RunID:      taskToken.RunID,
	})

	if !common.ValidIDLength(
		cancelRequest.GetIdentity(),
		scope,
		wh.config.MaxIDLengthWarnLimit(),
		wh.config.IdentityMaxLength(domainName),
		metrics.CadenceErrIdentityExceededWarnLimit,
		domainName,
		wh.GetLogger(),
		tag.IDTypeIdentity) {
		return wh.error(errIdentityTooLong, scope, tags...)
	}

	sizeLimitError := wh.config.BlobSizeLimitError(domainName)
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(domainName)

	if err := common.CheckEventBlobSizeLimit(
		len(cancelRequest.Details),
		sizeLimitWarn,
		sizeLimitError,
		taskToken.DomainID,
		taskToken.WorkflowID,
		taskToken.RunID,
		scope,
		wh.GetThrottledLogger(),
		tag.BlobSizeViolationOperation("RespondActivityTaskCanceled"),
	); err != nil {
		// details exceeds blob size limit, we would record it as failure
		failRequest := &types.RespondActivityTaskFailedRequest{
			TaskToken: cancelRequest.TaskToken,
			Reason:    common.StringPtr(common.FailureReasonCancelDetailsExceedsLimit),
			Details:   cancelRequest.Details[0:sizeLimitError],
			Identity:  cancelRequest.Identity,
		}
		err = wh.GetHistoryClient().RespondActivityTaskFailed(ctx, &types.HistoryRespondActivityTaskFailedRequest{
			DomainUUID:    taskToken.DomainID,
			FailedRequest: failRequest,
		})
		if err != nil {
			return wh.normalizeVersionedErrors(ctx, wh.error(err, scope, tags...))
		}
	} else {
		err = wh.GetHistoryClient().RespondActivityTaskCanceled(ctx, &types.HistoryRespondActivityTaskCanceledRequest{
			DomainUUID:    taskToken.DomainID,
			CancelRequest: cancelRequest,
		})
		if err != nil {
			return wh.normalizeVersionedErrors(ctx, wh.error(err, scope, tags...))
		}
	}

	return nil
}
// RespondActivityTaskCanceled - called to cancel an activity task
called to cancel an activity task
// RespondActivityTaskCanceled - called to cancel an activity task
func (wh *WorkflowHandler) RespondActivityTaskCanceled(
	ctx context.Context,
	cancelRequest *types.RespondActivityTaskCanceledRequest,
) (retError error) {
	defer log.CapturePanic(wh.GetLogger(), &retError)
	scope := wh.getDefaultScope(ctx, metrics.FrontendRespondActivityTaskCanceledScope)
	if wh.isShuttingDown() {
		return errShuttingDown
	}
	if err := wh.versionChecker.ClientSupported(ctx, wh.config.EnableClientVersionCheck()); err != nil {
		return wh.error(err, scope)
	}
	if cancelRequest == nil {
		return wh.error(errRequestNotSet, scope)
	}
	if cancelRequest.TaskToken == nil {
		return wh.error(errTaskTokenNotSet, scope)
	}
	taskToken, err := wh.tokenSerializer.Deserialize(cancelRequest.TaskToken)
	if err != nil {
		return wh.error(err, scope)
	}
	if taskToken.DomainID == "" {
		return wh.error(errDomainNotSet, scope)
	}
	domainName, err := wh.GetDomainCache().GetDomainName(taskToken.DomainID)
	if err != nil {
		return wh.error(err, scope)
	}
	dw := domainWrapper{
		domain: domainName,
	}
	scope, sw := wh.startRequestProfileWithDomain(
		ctx,
		metrics.FrontendRespondActivityTaskCanceledScope,
		dw,
	)
	defer sw.Stop()
	wh.allow(false, dw)
	tags := getDomainWfIDRunIDTags(domainName, &types.WorkflowExecution{
		WorkflowID: taskToken.WorkflowID,
		RunID:      taskToken.RunID,
	})
	if !common.ValidIDLength(
		cancelRequest.GetIdentity(),
		scope,
		wh.config.MaxIDLengthWarnLimit(),
		wh.config.IdentityMaxLength(domainName),
		metrics.CadenceErrIdentityExceededWarnLimit,
		domainName,
		wh.GetLogger(),
		tag.IDTypeIdentity) {
		return wh.error(errIdentityTooLong, scope, tags...)
	}
	sizeLimitError := wh.config.BlobSizeLimitError(domainName)
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(domainName)
	if err := common.CheckEventBlobSizeLimit(
		len(cancelRequest.Details),
		sizeLimitWarn,
		sizeLimitError,
		taskToken.DomainID,
		taskToken.WorkflowID,
		taskToken.RunID,
		scope,
		wh.GetThrottledLogger(),
		tag.BlobSizeViolationOperation("RespondActivityTaskCanceled"),
	); err != nil {
		failRequest := &types.RespondActivityTaskFailedRequest{
			TaskToken: cancelRequest.TaskToken,
			Reason:    common.StringPtr(common.FailureReasonCancelDetailsExceedsLimit),
			Details:   cancelRequest.Details[0:sizeLimitError],
			Identity:  cancelRequest.Identity,
		}
		err = wh.GetHistoryClient().RespondActivityTaskFailed(ctx, &types.HistoryRespondActivityTaskFailedRequest{
			DomainUUID:    taskToken.DomainID,
			FailedRequest: failRequest,
		})
		if err != nil {
			return wh.normalizeVersionedErrors(ctx, wh.error(err, scope, tags...))
		}
	} else {
		err = wh.GetHistoryClient().RespondActivityTaskCanceled(ctx, &types.HistoryRespondActivityTaskCanceledRequest{
			DomainUUID:    taskToken.DomainID,
			CancelRequest: cancelRequest,
		})
		if err != nil {
			return wh.normalizeVersionedErrors(ctx, wh.error(err, scope, tags...))
		}
	}
	return nil
}
func ( wh * WorkflowHandler ) RespondActivityTaskCanceled ( ctx context . Context , cancelRequest * types . RespondActivityTaskCanceledRequest , ) ( retError error ) { defer log . CapturePanic ( wh . GetLogger ( ) , & retError ) scope := wh . getDefaultScope ( ctx , metrics . FrontendRespondActivityTaskCanceledScope ) if wh . isShuttingDown ( ) { return errShuttingDown } if err := wh . versionChecker . ClientSupported ( ctx , wh . config . EnableClientVersionCheck ( ) ) ; err != nil { return wh . error ( err , scope ) } if cancelRequest == nil { return wh . error ( errRequestNotSet , scope ) } if cancelRequest . TaskToken == nil { return wh . error ( errTaskTokenNotSet , scope ) } taskToken , err := wh . tokenSerializer . Deserialize ( cancelRequest . TaskToken ) if err != nil { return wh . error ( err , scope ) } if taskToken . DomainID == " " { return wh . error ( errDomainNotSet , scope ) } domainName , err := wh . GetDomainCache ( ) . GetDomainName ( taskToken . DomainID ) if err != nil { return wh . error ( err , scope ) } dw := domainWrapper { domain : domainName , } scope , sw := wh . startRequestProfileWithDomain ( ctx , metrics . FrontendRespondActivityTaskCanceledScope , dw , ) defer sw . Stop ( ) wh . allow ( false , dw ) tags := getDomainWfIDRunIDTags ( domainName , & types . WorkflowExecution { WorkflowID : taskToken . WorkflowID , RunID : taskToken . RunID , } ) if ! common . ValidIDLength ( cancelRequest . GetIdentity ( ) , scope , wh . config . MaxIDLengthWarnLimit ( ) , wh . config . IdentityMaxLength ( domainName ) , metrics . CadenceErrIdentityExceededWarnLimit , domainName , wh . GetLogger ( ) , tag . IDTypeIdentity ) { return wh . error ( errIdentityTooLong , scope , tags ... ) } sizeLimitError := wh . config . BlobSizeLimitError ( domainName ) sizeLimitWarn := wh . config . BlobSizeLimitWarn ( domainName ) if err := common . CheckEventBlobSizeLimit ( len ( cancelRequest . Details ) , sizeLimitWarn , sizeLimitError , taskToken . DomainID , taskToken . WorkflowID , taskToken . RunID , scope , wh . GetThrottledLogger ( ) , tag . BlobSizeViolationOperation ( " " ) , ) ; err != nil { failRequest := & types . RespondActivityTaskFailedRequest { TaskToken : cancelRequest . TaskToken , Reason : common . StringPtr ( common . FailureReasonCancelDetailsExceedsLimit ) , Details : cancelRequest . Details [ 0 : sizeLimitError ] , Identity : cancelRequest . Identity , } err = wh . GetHistoryClient ( ) . RespondActivityTaskFailed ( ctx , & types . HistoryRespondActivityTaskFailedRequest { DomainUUID : taskToken . DomainID , FailedRequest : failRequest , } ) if err != nil { return wh . normalizeVersionedErrors ( ctx , wh . error ( err , scope , tags ... ) ) } } else { err = wh . GetHistoryClient ( ) . RespondActivityTaskCanceled ( ctx , & types . HistoryRespondActivityTaskCanceledRequest { DomainUUID : taskToken . DomainID , CancelRequest : cancelRequest , } ) if err != nil { return wh . normalizeVersionedErrors ( ctx , wh . error ( err , scope , tags ... ) ) } } return nil }
called to cancel an activity task
called to cancel an activity task
// Count the request in the host RPS, // but we still accept it even if RPS is exceeded // details exceeds blob size limit, we would record it as failure