Added standalone activity chasm dispatch task (#8540)

Added standalone activity Chasm tasks. Added handling of start activity
and e2e implementation of standalone activity start execution with
existing services. Updated protos related to standalone activities.

The Chasm tasks are needed to kick off standalone activity execution via
the existing services. Proto changes needed to so that the component ref
can be passed and handled via service stack.

- [X] built
- [X] run locally and tested manually
- [X] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)

---------

Co-authored-by: Roey Berman <[email protected]>
Co-authored-by: Dan Davison <[email protected]>

Author: 3 people

api/historyservice/v1/request_response.pb.go

@@ -2,18 +2,26 @@ package activity
 
 import (
 	commonpb "go.temporal.io/api/common/v1"
+	deploymentpb "go.temporal.io/api/deployment/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	historypb "go.temporal.io/api/history/v1"
+	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/server/api/historyservice/v1"
+	"go.temporal.io/server/api/matchingservice/v1"
+	taskqueuespb "go.temporal.io/server/api/taskqueue/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 type ActivityStore interface {
-	PopulateRecordActivityTaskStartedResponse(ctx chasm.Context, res *historyservice.RecordActivityTaskStartedResponse) error
+	PopulateRecordActivityTaskStartedResponse(ctx chasm.Context, key chasm.EntityKey, response *historyservice.RecordActivityTaskStartedResponse) error
 	RecordCompletion(ctx chasm.MutableContext) error
 }
 
+// Activity component represents an activity execution persistence object and can be either standalone activity or one
+// embedded within a workflow.
 type Activity struct {
 	chasm.UnimplementedComponent
 
@@ -26,14 +34,21 @@ type Activity struct {
 	// Standalone only
 	RequestData chasm.Field[*activitypb.ActivityRequestData]
 
-	// Pointer to an implementation of the "store" (for a workflow activity this would be a parent pointer back to
-	// the workflow).
+	// Pointer to an implementation of the "store". for a workflow activity this would be a parent pointer back to
+	// the workflow. For a standalone activity this would be nil.
+	// TODO: revisit a standalone activity pointing to itself once we handle storing it more efficiently.
 	// TODO: figure out better naming.
 	Store chasm.Field[ActivityStore]
 }
 
+// RecordActivityTaskStartedParams holds parameters for RecordActivityTaskStarted
+type RecordActivityTaskStartedParams struct {
+	VersionDirective *taskqueuespb.TaskVersionDirective
+	WorkerIdentity   string
+}
+
 // LifecycleState TODO: we need to add more lifecycle states to better categorize some activity states, particulary for terminated/canceled.
-func (a Activity) LifecycleState(_ chasm.Context) chasm.LifecycleState {
+func (a *Activity) LifecycleState(_ chasm.Context) chasm.LifecycleState {
 	switch a.Status {
 	case activitypb.ACTIVITY_EXECUTION_STATUS_COMPLETED:
 		return chasm.LifecycleStateCompleted
@@ -47,6 +62,7 @@ func (a Activity) LifecycleState(_ chasm.Context) chasm.LifecycleState {
 	}
 }
 
+// NewStandaloneActivity creates a new activity component and adds associated tasks to start execution.
 func NewStandaloneActivity(
 	ctx chasm.MutableContext,
 	request *workflowservice.StartActivityExecutionRequest,
@@ -56,23 +72,32 @@ func NewStandaloneActivity(
 		return nil, err
 	}
 
-	return &Activity{
+	// TODO flatten this when API is updated
+	options := request.GetOptions()
+
+	activity := &Activity{
 		ActivityState: &activitypb.ActivityState{
-			ActivityType:    request.ActivityType,
-			ActivityOptions: request.Options,
-			Priority:        request.Priority,
+			ActivityType:           request.ActivityType,
+			TaskQueue:              options.GetTaskQueue(),
+			ScheduleToCloseTimeout: options.GetScheduleToCloseTimeout(),
+			ScheduleToStartTimeout: options.GetScheduleToStartTimeout(),
+			StartToCloseTimeout:    options.GetStartToCloseTimeout(),
+			HeartbeatTimeout:       options.GetHeartbeatTimeout(),
+			RetryPolicy:            options.GetRetryPolicy(),
+			Priority:               request.Priority,
 		},
 		Attempt: chasm.NewDataField(ctx, &activitypb.ActivityAttemptState{
-			Count:           1,
-			LastStartedTime: timestamppb.Now(),
+			Count: 1,
 		}),
 		RequestData: chasm.NewDataField(ctx, &activitypb.ActivityRequestData{
 			Input:        request.Input,
 			Header:       request.Header,
 			UserMetadata: request.UserMetadata,
 		}),
 		Visibility: chasm.NewComponentField(ctx, visibility),
-	}, nil
+	}
+
+	return activity, nil
 }
 
 func NewEmbeddedActivity(
@@ -82,18 +107,121 @@ func NewEmbeddedActivity(
 ) {
 }
 
-func (a *Activity) PopulateRecordActivityTaskStartedResponse(ctx chasm.Context, res *historyservice.RecordActivityTaskStartedResponse) error {
+func (a *Activity) createAddActivityTaskRequest(ctx chasm.Context, namespaceID string) (*matchingservice.AddActivityTaskRequest, error) {
+	// Get latest component ref and unmarshal into proto ref
+	componentRef, err := ctx.Ref(a)
+	if err != nil {
+		return nil, err
+	}
+
+	// Note: No need to set the vector clock here, as the components track version conflicts for read/write
+	return &matchingservice.AddActivityTaskRequest{
+		NamespaceId:            namespaceID,
+		TaskQueue:              a.GetTaskQueue(),
+		ScheduleToStartTimeout: a.GetScheduleToStartTimeout(),
+		Priority:               a.GetPriority(),
+		ComponentRef:           componentRef,
+	}, nil
+}
+
+// RecordActivityTaskStarted updates the activity on recording activity task started and populates the response.
+func (a *Activity) RecordActivityTaskStarted(ctx chasm.MutableContext, params RecordActivityTaskStartedParams) (*historyservice.RecordActivityTaskStartedResponse, error) {
+	if err := TransitionStarted.Apply(a, ctx, nil); err != nil {
+		return nil, err
+	}
+
+	attempt, err := a.Attempt.Get(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	attempt.LastStartedTime = timestamppb.New(ctx.Now(a))
+	attempt.LastWorkerIdentity = params.WorkerIdentity
+
+	if versionDirective := params.VersionDirective.GetDeploymentVersion(); versionDirective != nil {
+		attempt.LastDeploymentVersion = &deploymentpb.WorkerDeploymentVersion{
+			BuildId:        versionDirective.GetBuildId(),
+			DeploymentName: versionDirective.GetDeploymentName(),
+		}
+	}
+
 	store, err := a.Store.Get(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	activityRefBytes, err := ctx.Ref(a)
+	if err != nil {
+		return nil, err
+	}
+
+	activityRef, err := chasm.DeserializeComponentRef(activityRefBytes)
+	if err != nil {
+		return nil, err
+	}
+
+	response := &historyservice.RecordActivityTaskStartedResponse{}
+	if store == nil {
+		// TODO Get entity key from from context once we rebase on main
+		if err := a.PopulateRecordActivityTaskStartedResponse(ctx, activityRef.EntityKey, response); err != nil {
+			return nil, err
+		}
+	} else {
+		if err := store.PopulateRecordActivityTaskStartedResponse(ctx, activityRef.EntityKey, response); err != nil {
+			return nil, err
+		}
+	}
+
+	return response, nil
+}
+
+func (a *Activity) PopulateRecordActivityTaskStartedResponse(ctx chasm.Context, key chasm.EntityKey, response *historyservice.RecordActivityTaskStartedResponse) error {
+	attempt, err := a.Attempt.Get(ctx)
 	if err != nil {
 		return err
 	}
-	if err := store.PopulateRecordActivityTaskStartedResponse(ctx, res); err != nil {
+
+	lastHeartbeat, err := a.LastHeartbeat.Get(ctx)
+	if err != nil {
 		return err
 	}
-	// ...
+
+	requestData, err := a.RequestData.Get(ctx)
+	if err != nil {
+		return err
+	}
+
+	response.StartedTime = attempt.LastStartedTime
+	response.Attempt = attempt.GetCount()
+	if lastHeartbeat != nil {
+		response.HeartbeatDetails = lastHeartbeat.GetDetails()
+	}
+	response.Priority = a.GetPriority()
+	response.RetryPolicy = a.GetRetryPolicy()
+	response.ScheduledEvent = &historypb.HistoryEvent{
+		EventType: enumspb.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED,
+		Attributes: &historypb.HistoryEvent_ActivityTaskScheduledEventAttributes{
+			ActivityTaskScheduledEventAttributes: &historypb.ActivityTaskScheduledEventAttributes{
+				ActivityId:             key.BusinessID,
+				ActivityType:           a.GetActivityType(),
+				Input:                  requestData.GetInput(),
+				Header:                 requestData.GetHeader(),
+				TaskQueue:              a.GetTaskQueue(),
+				ScheduleToCloseTimeout: a.GetScheduleToCloseTimeout(),
+				ScheduleToStartTimeout: a.GetScheduleToStartTimeout(),
+				StartToCloseTimeout:    a.GetStartToCloseTimeout(),
+				HeartbeatTimeout:       a.GetHeartbeatTimeout(),
+			},
+		},
+	}
+
 	return nil
 }
 
+func (a *Activity) RecordCompletion(_ chasm.MutableContext) error {
+	return serviceerror.NewUnimplemented("RecordCompletion is not implemented")
+}
+
 func (a *Activity) RecordHeartbeat(ctx chasm.MutableContext, details *commonpb.Payloads) (chasm.NoValue, error) {
 	a.LastHeartbeat = chasm.NewDataField(ctx, &activitypb.ActivityHeartbeatState{
 		RecordedTime: timestamppb.New(ctx.Now(a)),