Add support for standalone activity cancellation

chasm/lib/activity/activity.go

@@ -259,6 +259,16 @@ func (a *Activity) HandleFailed(ctx chasm.MutableContext, req *historyservice.Re
 	return &historyservice.RespondActivityTaskFailedResponse{}, nil
 }
 
+// HandleCanceled updates the activity on activity canceled.
+func (a *Activity) HandleCanceled(ctx chasm.MutableContext, request *historyservice.RespondActivityTaskCanceledRequest) (
+	*historyservice.RespondActivityTaskCanceledResponse, error) {
+	if err := TransitionCanceled.Apply(a, ctx, request); err != nil {
+		return nil, err
+	}
+
+	return &historyservice.RespondActivityTaskCanceledResponse{}, nil
+}
+
 func (a *Activity) handleTerminated(ctx chasm.MutableContext, req *activitypb.TerminateActivityExecutionRequest) (
 	*activitypb.TerminateActivityExecutionResponse, error) {
 	if err := TransitionTerminated.Apply(a, ctx, req); err != nil {
@@ -284,6 +294,28 @@ func (a *Activity) getLastHeartbeat(ctx chasm.MutableContext) (*activitypb.Activ
 	return heartbeat, nil
 }
 
+func (a *Activity) handleCancellationRequested(ctx chasm.MutableContext, req *activitypb.CancelActivityExecutionRequest) (
+	*activitypb.CancelActivityExecutionResponse, error) {
+	newReqID := req.GetFrontendRequest().GetRequestId()
+	existingReqID := a.GetCancelState().GetRequestId()
+
+	// If already in cancel requested state, fail if request ID is different, else no-op
+	if a.ActivityState.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED {
+		if existingReqID != newReqID {
+			return nil, serviceerror.NewFailedPrecondition(
+				fmt.Sprintf("cancellation already requested with request ID %s", existingReqID))
+		}
+
+		return &activitypb.CancelActivityExecutionResponse{}, nil
+	}
+
+	if err := TransitionCancelRequested.Apply(a, ctx, req); err != nil {
+		return nil, err
+	}
+
+	return &activitypb.CancelActivityExecutionResponse{}, nil
+}
+
 func (a *Activity) shouldRetryOnFailure(ctx chasm.Context, failure *failurepb.Failure) (bool, time.Duration, error) {
 	var isRetryable bool
 
@@ -370,6 +402,10 @@ func (a *Activity) recordFailedAttempt(
 }
 
 func (a *Activity) shouldRetry(ctx chasm.Context, overridingRetryInterval time.Duration) (bool, time.Duration, error) {
+	if !TransitionRescheduled.Possible(a) {
+		return false, 0, nil
+	}
+
 	attempt, err := a.Attempt.Get(ctx)
 	if err != nil {
 		return false, 0, err
@@ -484,6 +520,7 @@ func (a *Activity) buildActivityExecutionInfo(ctx chasm.Context) (*activity.Acti
 		ActivityId:              key.BusinessID,
 		ActivityType:            a.GetActivityType(),
 		Attempt:                 attempt.GetCount(),
+		CanceledReason:          a.CancelState.GetReason(),
 		Header:                  requestData.GetHeader(),
 		HeartbeatDetails:        heartbeat.GetDetails(),
 		LastAttemptCompleteTime: attempt.GetCompleteTime(),

chasm/lib/activity/frontend.go

@@ -131,6 +131,33 @@ func (h *frontendHandler) TerminateActivityExecution(
 	return &workflowservice.TerminateActivityExecutionResponse{}, nil
 }
 
+func (h *frontendHandler) RequestCancelActivityExecution(
+	ctx context.Context,
+	req *workflowservice.RequestCancelActivityExecutionRequest,
+) (*workflowservice.RequestCancelActivityExecutionResponse, error) {
+	namespaceID, err := h.namespaceRegistry.GetNamespaceID(namespace.Name(req.GetNamespace()))
+	if err != nil {
+		return nil, err
+	}
+
+	// Since validation potentially mutates the request, we clone it first so that any retries use the original request.
+	req = common.CloneProto(req)
+	err = validateAndNormalizeRequestID(&req.RequestId, dynamicconfig.MaxIDLengthLimit.Get(h.dc)())
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = h.client.CancelActivityExecution(ctx, &activitypb.CancelActivityExecutionRequest{
+		NamespaceId:     namespaceID.String(),
+		FrontendRequest: req,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &workflowservice.RequestCancelActivityExecutionResponse{}, nil
+}
+
 func (h *frontendHandler) validateAndPopulateStartRequest(
 	req *workflowservice.StartActivityExecutionRequest,
 	namespaceID namespace.ID,