PollComponent and PollActivityExecution

chasm/context.go

@@ -21,6 +21,7 @@ type Context interface {
 	// Intent() OperationIntent
 	// ComponentOptions(Component) []ComponentOption
 
+	structuredRef(Component) (ComponentRef, error)
 	getContext() context.Context
 }
 
@@ -94,6 +95,10 @@ func (c *immutableCtx) Ref(component Component) ([]byte, error) {
 	return c.root.Ref(component)
 }
 
+func (c *immutableCtx) structuredRef(component Component) (ComponentRef, error) {
+	return c.root.structuredRef(component)
+}
+
 func (c *immutableCtx) Now(component Component) time.Time {
 	return c.root.Now(component)
 }

chasm/context_mock.go

@@ -8,9 +8,10 @@ import (
 
 // MockContext is a mock implementation of [Context].
 type MockContext struct {
-	HandleExecutionKey func() EntityKey
-	HandleNow          func(component Component) time.Time
-	HandleRef          func(component Component) ([]byte, error)
+	HandleExecutionKey  func() EntityKey
+	HandleNow           func(component Component) time.Time
+	HandleRef           func(component Component) ([]byte, error)
+	HandleStructuredRef func(component Component) (ComponentRef, error)
 }
 
 func (c *MockContext) getContext() context.Context {
@@ -31,6 +32,13 @@ func (c *MockContext) Ref(cmp Component) ([]byte, error) {
 	return nil, nil
 }
 
+func (c *MockContext) structuredRef(cmp Component) (ComponentRef, error) {
+	if c.HandleStructuredRef != nil {
+		return c.HandleStructuredRef(cmp)
+	}
+	return ComponentRef{}, nil
+}
+
 func (c *MockContext) ExecutionKey() EntityKey {
 	if c.HandleExecutionKey != nil {
 		return c.HandleExecutionKey()

chasm/engine.go

@@ -42,10 +42,12 @@ type Engine interface {
 	PollComponent(
 		context.Context,
 		ComponentRef,
-		func(Context, Component) (any, bool, error),
-		func(MutableContext, Component, any) error,
+		func(Context, Component) (bool, error),
 		...TransitionOption,
 	) ([]byte, error)
+
+	// NotifyExecution notifies any PollComponent callers waiting on the execution.
+	NotifyExecution(EntityKey)
 }
 
 type BusinessIDReusePolicy int
@@ -176,6 +178,9 @@ func UpdateWithNewEntity[C Component, I any, O1 any, O2 any](
 //   - consider remove ComponentRef from the return value and allow components to get
 //     the ref in the transition function. There are some caveats there, check the
 //     comment of the NewRef method in MutableContext.
+//
+// UpdateComponent applies updateFn to the component identified by the supplied component reference.
+// It returns the result, along with the new component reference. opts are currently ignored.
 func UpdateComponent[C Component, R []byte | ComponentRef, I any, O any](
 	ctx context.Context,
 	r R,
@@ -207,6 +212,8 @@ func UpdateComponent[C Component, R []byte | ComponentRef, I any, O any](
 	return output, newSerializedRef, err
 }
 
+// ReadComponent returns the result of evaluating readFn against the component identified by the
+// component reference. opts are currently ignored.
 func ReadComponent[C Component, R []byte | ComponentRef, I any, O any](
 	ctx context.Context,
 	r R,
@@ -234,18 +241,18 @@ func ReadComponent[C Component, R []byte | ComponentRef, I any, O any](
 	return output, err
 }
 
-type PollComponentRequest[C Component, I any, O any] struct {
-	Ref         ComponentRef
-	PredicateFn func(C, Context, I) bool
-	OperationFn func(C, MutableContext, I) (O, error)
-	Input       I
-}
-
-func PollComponent[C Component, R []byte | ComponentRef, I any, O any, T any](
+// PollComponent waits until the predicate is true when evaluated against the component identified
+// by the supplied component reference. If this times out due to a server-imposed long-poll timeout
+// then it returns (nil, nil, nil), as an indication that the caller should continue long-polling.
+// Otherwise it returns (output, ref, err), where output is the output of the predicate function,
+// and ref is a component reference identifying the state at which the predicate was satisfied. The
+// predicate must be monotonic: if it returns true at execution state transition s then it must
+// return true at all transitions t > s. If the predicate is true at the outset then PollComponent
+// returns immediately. opts are currently ignored.
+func PollComponent[C Component, R []byte | ComponentRef, I any, O any](
 	ctx context.Context,
 	r R,
-	predicateFn func(C, Context, I) (T, bool, error),
-	operationFn func(C, MutableContext, I, T) (O, error),
+	monotonicPredicate func(C, Context, I) (O, bool, error),
 	input I,
 	opts ...TransitionOption,
 ) (O, []byte, error) {
@@ -259,13 +266,12 @@ func PollComponent[C Component, R []byte | ComponentRef, I any, O any, T any](
 	newSerializedRef, err := engineFromContext(ctx).PollComponent(
 		ctx,
 		ref,
-		func(ctx Context, c Component) (any, bool, error) {
-			return predicateFn(c.(C), ctx, input)
-		},
-		func(ctx MutableContext, c Component, t any) error {
-			var err error
-			output, err = operationFn(c.(C), ctx, input, t.(T))
-			return err
+		func(ctx Context, c Component) (bool, error) {
+			out, satisfied, err := monotonicPredicate(c.(C), ctx, input)
+			if satisfied {
+				output = out
+			}
+			return satisfied, err
 		},
 		opts...,
 	)

chasm/lib/activity/activity.go

@@ -4,11 +4,13 @@ import (
 	"fmt"
 	"time"
 
+	"go.temporal.io/api/activity/v1"
 	commonpb "go.temporal.io/api/common/v1"
 	deploymentpb "go.temporal.io/api/deployment/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	failurepb "go.temporal.io/api/failure/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"
@@ -37,14 +39,12 @@ type Activity struct {
 
 	*activitypb.ActivityState
 
-	// Standalone only
 	Visibility    chasm.Field[*chasm.Visibility]
 	Attempt       chasm.Field[*activitypb.ActivityAttemptState]
 	LastHeartbeat chasm.Field[*activitypb.ActivityHeartbeatState]
 	// Standalone only
 	RequestData chasm.Field[*activitypb.ActivityRequestData]
 	Outcome     chasm.Field[*activitypb.ActivityOutcome]
-
 	// 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.
@@ -314,3 +314,141 @@ func (a *Activity) RecordHeartbeat(ctx chasm.MutableContext, details *commonpb.P
 	})
 	return nil, nil
 }
+
+func (a *Activity) buildActivityExecutionInfo(ctx chasm.Context) (*activity.ActivityExecutionInfo, error) {
+	// TODO(dan): support pause states
+	var status enumspb.ActivityExecutionStatus
+	var runState enumspb.PendingActivityState
+	switch a.GetStatus() {
+	case activitypb.ACTIVITY_EXECUTION_STATUS_UNSPECIFIED:
+		status = enumspb.ACTIVITY_EXECUTION_STATUS_UNSPECIFIED
+		runState = enumspb.PENDING_ACTIVITY_STATE_UNSPECIFIED
+	case activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED:
+		status = enumspb.ACTIVITY_EXECUTION_STATUS_RUNNING
+		runState = enumspb.PENDING_ACTIVITY_STATE_SCHEDULED
+	case activitypb.ACTIVITY_EXECUTION_STATUS_STARTED:
+		status = enumspb.ACTIVITY_EXECUTION_STATUS_RUNNING
+		runState = enumspb.PENDING_ACTIVITY_STATE_STARTED
+	case activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED:
+		status = enumspb.ACTIVITY_EXECUTION_STATUS_RUNNING
+		runState = enumspb.PENDING_ACTIVITY_STATE_CANCEL_REQUESTED
+	case activitypb.ACTIVITY_EXECUTION_STATUS_COMPLETED:
+		status = enumspb.ACTIVITY_EXECUTION_STATUS_COMPLETED
+		runState = enumspb.PENDING_ACTIVITY_STATE_UNSPECIFIED
+	case activitypb.ACTIVITY_EXECUTION_STATUS_FAILED:
+		status = enumspb.ACTIVITY_EXECUTION_STATUS_FAILED
+		runState = enumspb.PENDING_ACTIVITY_STATE_UNSPECIFIED
+	case activitypb.ACTIVITY_EXECUTION_STATUS_CANCELED:
+		status = enumspb.ACTIVITY_EXECUTION_STATUS_CANCELED
+		runState = enumspb.PENDING_ACTIVITY_STATE_UNSPECIFIED
+	case activitypb.ACTIVITY_EXECUTION_STATUS_TERMINATED:
+		status = enumspb.ACTIVITY_EXECUTION_STATUS_TERMINATED
+		runState = enumspb.PENDING_ACTIVITY_STATE_UNSPECIFIED
+	case activitypb.ACTIVITY_EXECUTION_STATUS_TIMED_OUT:
+		status = enumspb.ACTIVITY_EXECUTION_STATUS_TIMED_OUT
+		runState = enumspb.PENDING_ACTIVITY_STATE_UNSPECIFIED
+	default:
+		return nil, serviceerror.NewInternalf("unknown activity execution status: %s", a.GetStatus())
+	}
+
+	requestData, err := a.RequestData.Get(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	attempt, err := a.Attempt.Get(ctx)
+	if err != nil {
+		return nil, err
+	}
+	key := ctx.ExecutionKey()
+
+	info := &activity.ActivityExecutionInfo{
+		ActivityId:    key.BusinessID,
+		RunId:         key.EntityID,
+		ActivityType:  a.GetActivityType(),
+		Status:        status,
+		RunState:      runState,
+		ScheduledTime: a.GetScheduledTime(),
+		Priority:      a.GetPriority(),
+		Header:        requestData.GetHeader(),
+		LastFailure:   attempt.GetLastFailureDetails().GetFailure(),
+		// TODO(dan): populate remaining fields
+	}
+
+	return info, nil
+}
+
+func (a *Activity) buildPollActivityExecutionResponse(
+	ctx chasm.Context,
+	req *activitypb.PollActivityExecutionRequest,
+) (*activitypb.PollActivityExecutionResponse, error) {
+	request := req.GetFrontendRequest()
+
+	token, err := ctx.Ref(a)
+	if err != nil {
+		return nil, err
+	}
+
+	var info *activity.ActivityExecutionInfo
+	if request.GetIncludeInfo() {
+		info, err = a.buildActivityExecutionInfo(ctx)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	var input *commonpb.Payloads
+	if request.GetIncludeInput() {
+		activityRequest, err := a.RequestData.Get(ctx)
+		if err != nil {
+			return nil, err
+		}
+		input = activityRequest.GetInput()
+	}
+
+	response := &workflowservice.PollActivityExecutionResponse{
+		Info:                     info,
+		RunId:                    ctx.ExecutionKey().EntityID,
+		Input:                    input,
+		StateChangeLongPollToken: token,
+	}
+
+	if request.GetIncludeOutcome() {
+		activityOutcome, err := a.Outcome.Get(ctx)
+		if err != nil {
+			return nil, err
+		}
+		// There are two places where a failure might be stored but only one place where a
+		// successful outcome is stored.
+		if successful := activityOutcome.GetSuccessful(); successful != nil {
+			response.Outcome = &workflowservice.PollActivityExecutionResponse_Result{
+				Result: successful.GetOutput(),
+			}
+		} else if failure := activityOutcome.GetFailed().GetFailure(); failure != nil {
+			response.Outcome = &workflowservice.PollActivityExecutionResponse_Failure{
+				Failure: failure,
+			}
+		} else {
+			shouldHaveFailure := (a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_FAILED ||
+				a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_TIMED_OUT ||
+				a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_CANCELED ||
+				a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_TERMINATED)
+
+			if shouldHaveFailure {
+				attempt, err := a.Attempt.Get(ctx)
+				if err != nil {
+					return nil, err
+				}
+				if details := attempt.GetLastFailureDetails(); details != nil {
+					response.Outcome = &workflowservice.PollActivityExecutionResponse_Failure{
+						Failure: details.GetFailure(),
+					}
+				}
+			}
+		}
+	}
+
+	return &activitypb.PollActivityExecutionResponse{
+		FrontendResponse: response,
+	}, nil
+}

chasm/lib/activity/config.go

@@ -0,0 +1,35 @@
+package activity
+
+import (
+	"time"
+
+	"go.temporal.io/server/common/dynamicconfig"
+)
+
+var (
+	LongPollTimeout = dynamicconfig.NewNamespaceDurationSetting(
+		"chasm.activity.longPollTimeout",
+		20*time.Second,
+		`Timeout for activity long-poll requests.`,
+	)
+
+	LongPollBuffer = dynamicconfig.NewNamespaceDurationSetting(
+		"chasm.activity.longPollBuffer",
+		time.Second,
+		`A buffer used to adjust the activity long-poll timeouts.
+ Specifically, activity long-poll requests are timed out at a time which leaves at least the buffer's duration
+ remaining before the caller's deadline, if permitted by the caller's deadline.`,
+	)
+)
+
+type Config struct {
+	LongPollTimeout dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	LongPollBuffer  dynamicconfig.DurationPropertyFnWithNamespaceFilter
+}
+
+func ConfigProvider(dc *dynamicconfig.Collection) *Config {
+	return &Config{
+		LongPollTimeout: LongPollTimeout.Get(dc),
+		LongPollBuffer:  LongPollBuffer.Get(dc),
+	}
+}