Standalone Activity: DescribeActivityExecution and GetActivityExecutionOutcome #8771
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -76,21 +76,18 @@ func NewStandaloneActivity( | |
| visibility := chasm.NewVisibilityWithData( | ||
| ctx, | ||
| request.GetSearchAttributes().GetIndexedFields(), | ||
| nil, | ||
| ) | ||
|
|
||
| activity := &Activity{ | ||
| ActivityState: &activitypb.ActivityState{ | ||
| ActivityType: request.ActivityType, | ||
| TaskQueue: request.GetTaskQueue(), | ||
| ScheduleToCloseTimeout: request.GetScheduleToCloseTimeout(), | ||
| ScheduleToStartTimeout: request.GetScheduleToStartTimeout(), | ||
| StartToCloseTimeout: request.GetStartToCloseTimeout(), | ||
| HeartbeatTimeout: request.GetHeartbeatTimeout(), | ||
| RetryPolicy: request.GetRetryPolicy(), | ||
| Priority: request.Priority, | ||
| }, | ||
| LastAttempt: chasm.NewDataField(ctx, &activitypb.ActivityAttemptState{}), | ||
|
|
@@ -486,82 +483,106 @@ func (a *Activity) buildActivityExecutionInfo(ctx chasm.Context) (*activity.Acti | |
| Priority: a.GetPriority(), | ||
| RunId: key.RunID, | ||
| RunState: runState, | ||
| ScheduleTime: a.GetScheduledTime(), | ||
| Status: status, | ||
| // TODO(dan): populate remaining fields | ||
| } | ||
|
|
||
| return info, nil | ||
| } | ||
|
|
||
| func (a *Activity) buildDescribeActivityExecutionResponse( | ||
| ctx chasm.Context, | ||
| req *activitypb.DescribeActivityExecutionRequest, | ||
| ) (*activitypb.DescribeActivityExecutionResponse, error) { | ||
| request := req.GetFrontendRequest() | ||
|
|
||
| token, err := ctx.Ref(a) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| info, err := a.buildActivityExecutionInfo(ctx) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| var input *commonpb.Payloads | ||
| if request.GetIncludeInput() { | ||
| input = a.RequestData.Get(ctx).GetInput() | ||
| } | ||
|
|
||
| response := &workflowservice.DescribeActivityExecutionResponse{ | ||
| Info: info, | ||
| RunId: ctx.ExecutionKey().RunID, | ||
| Input: input, | ||
| LongPollToken: token, | ||
| } | ||
|
|
||
| if request.GetIncludeOutcome() { | ||
| result, failure, err := a.getOutcome(ctx) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| if result != nil { | ||
| response.Outcome = &workflowservice.DescribeActivityExecutionResponse_Result{Result: result} | ||
|
||
| } else if failure != nil { | ||
| response.Outcome = &workflowservice.DescribeActivityExecutionResponse_Failure{Failure: failure} | ||
| } | ||
| } | ||
|
|
||
| return &activitypb.DescribeActivityExecutionResponse{ | ||
| FrontendResponse: response, | ||
| }, nil | ||
| } | ||
|
|
||
| func (a *Activity) buildGetActivityExecutionOutcomeResponse( | ||
| ctx chasm.Context, | ||
| ) (*activitypb.GetActivityExecutionOutcomeResponse, error) { | ||
| result, failure, err := a.getOutcome(ctx) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| response := &workflowservice.GetActivityExecutionOutcomeResponse{ | ||
| RunId: ctx.ExecutionKey().RunID, | ||
| } | ||
| if result != nil { | ||
| response.Outcome = &workflowservice.GetActivityExecutionOutcomeResponse_Result{Result: result} | ||
| } else if failure != nil { | ||
| response.Outcome = &workflowservice.GetActivityExecutionOutcomeResponse_Failure{Failure: failure} | ||
| } | ||
| return &activitypb.GetActivityExecutionOutcomeResponse{ | ||
| FrontendResponse: response, | ||
| }, nil | ||
| } | ||
|
|
||
| // getOutcome retrieves the activity outcome (result or failure) if the activity has completed. | ||
| // Returns (result, failure, error) where at most one of result/failure is non-nil. | ||
| func (a *Activity) getOutcome(ctx chasm.Context) (*commonpb.Payloads, *failurepb.Failure, error) { | ||
|
||
| activityOutcome := a.Outcome.Get(ctx) | ||
| // Check for successful outcome | ||
| if successful := activityOutcome.GetSuccessful(); successful != nil { | ||
| return successful.GetOutput(), nil, nil | ||
| } | ||
| // Check for failure in outcome | ||
| if failure := activityOutcome.GetFailed().GetFailure(); failure != nil { | ||
|
Member
There was a problem hiding this comment. Please verify that we don't set failure to an empty struct when we fail from an attempt.
Contributor
Author
There was a problem hiding this comment. We were still doing that (I'd brought it up before but we hadn't removed it). I removed it in 32cd5cf (PTAL). |
||
| return nil, failure, nil | ||
| } | ||
| // Check for failure in last attempt details | ||
| shouldHaveFailure := (a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_FAILED || | ||
|
Member
There was a problem hiding this comment. Just noticed this: terminated and timed out should never return outcome from the last attempt, canceled would sometimes have to read the last attempt failure.
Member
There was a problem hiding this comment. Might be best to first check if the lifecycle state is not running at the top of this function and then go through, outcome success, outcome failure, and last result without checking statuses here.
Contributor
Author
There was a problem hiding this comment. I've moved this to an issue to track because I want to write a test to repro it and we can unblock others by merging this now. |
||
| a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_TIMED_OUT || | ||
| a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_CANCELED || | ||
| a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_TERMINATED) | ||
| if shouldHaveFailure { | ||
| if details := a.LastAttempt.Get(ctx).GetLastFailureDetails(); details != nil { | ||
| return nil, details.GetFailure(), nil | ||
| } | ||
|
There was a problem hiding this comment. Bug: Incorrect outcome returned for terminated/timed-out activitiesThe |
||
| } | ||
| return nil, nil, nil | ||
| } | ||
|
|
||
| // StoreOrSelf returns the store for the activity. If the store is not set as a field (e.g. | ||
| // standalone activities), it returns the activity itself. | ||
| func (a *Activity) StoreOrSelf(ctx chasm.MutableContext) ActivityStore { | ||
| store, ok := a.Store.TryGet(ctx) | ||
| if ok { | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Might as well rename to
schedule_timein the internal protos too.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done