Add CheckPrerequisite to WorkflowStep to validate state machine edges#374
Add CheckPrerequisite to WorkflowStep to validate state machine edges#374Zoe Zhao (zoez7) wants to merge 4 commits into
Conversation
Zoe Zhao (zoez7)
force-pushed
the
check-prerequisite
branch
4 times, most recently
from
9e8acdf to
3921ac8
Compare
| } | ||
| func (s *AssignWorkerStep) CheckPrerequisite(ctx context.Context, input *ResumeInput, state *ResumeState) error { | ||
| // The resume edge exists from SUSPENDED and PAUSED. | ||
| // RESUMING is allowed for retrying this step. |
There was a problem hiding this comment.
This is because of line 168 below.
There was a problem hiding this comment.
could you please clarify what is special at line 168? Is it for cases it was an error in resuming process and we are OK to try resuming one more time?
Same case might happening with Pausing and the CheckPrerequisite for CallAteletPauseStep does not accept pausing state.
There was a problem hiding this comment.
I updated the logic here and stopped allowing resuming state to reenter the AssignWorkerStep. Because setting the state to Resuming is the last step of AssignWorkerStep, if the actor is already in resuming, we know it has completed all the steps and don't need to retry.
On line 168, we would check which worker is assigned and if the worker is no longer eligible, we will pick another one, but this is for cleaning up and doesn't have to be retried during this operation.
Zoe Zhao (zoez7)
force-pushed
the
check-prerequisite
branch
from
3921ac8 to
ec3b70c
Compare
Zoe Zhao (zoez7)
force-pushed
the
check-prerequisite
branch
from
ec3b70c to
4104a7d
Compare
github-project-automation
Bot
moved this to Todo
in Substrate Milestones (go/substrate-milestones)
| return nil | ||
| } | ||
| func (s *CallAteletPauseStep) Execute(ctx context.Context, input *PauseInput, state *PauseState) error { | ||
| if state.Actor.GetAteomPodNamespace() == "" || state.Actor.GetAteomPodName() == "" { |
There was a problem hiding this comment.
this is a pre-requisite too, however function in current implementation does not support transition to crashActor.
Might be rename the "CheckPrerequisite" to different name? might be allow transition to crash? or might be from the beginning we were not supposed to be in this state, that actor does not have pod or namespace?
There was a problem hiding this comment.
We should crash the actor here. We can only get here if MarkPausing has succeeded, not having the ateom pod or namespace should crash the actor. I changed the implementation.
| return nil | ||
| } | ||
| func (s *CallAteletPauseStep) Execute(ctx context.Context, input *PauseInput, state *PauseState) error { | ||
| if state.Actor.GetAteomPodNamespace() == "" || state.Actor.GetAteomPodName() == "" { |
There was a problem hiding this comment.
We should crash the actor here. We can only get here if MarkPausing has succeeded, not having the ateom pod or namespace should crash the actor. I changed the implementation.
| } | ||
| func (s *AssignWorkerStep) CheckPrerequisite(ctx context.Context, input *ResumeInput, state *ResumeState) error { | ||
| // The resume edge exists from SUSPENDED and PAUSED. | ||
| // RESUMING is allowed for retrying this step. |
There was a problem hiding this comment.
I updated the logic here and stopped allowing resuming state to reenter the AssignWorkerStep. Because setting the state to Resuming is the last step of AssignWorkerStep, if the actor is already in resuming, we know it has completed all the steps and don't need to retry.
On line 168, we would check which worker is assigned and if the worker is no longer eligible, we will pick another one, but this is for cleaning up and doesn't have to be retried during this operation.
| state.ActorTemplate = actorTemplate | ||
|
|
||
| // If the Actor is in Resuming state, it means a previous attempt crashed after AssignWorkerStep. | ||
| // We don't need to repeat the AssignWorkerStep, load the Worker now. |
There was a problem hiding this comment.
I added this step so if we retry resumeActor for Actor already in RESUMING state, it will still load worker to the ResumeState
| func (s *CallAteletRestoreStep) IsComplete(ctx context.Context, input *ResumeInput, state *ResumeState) (bool, error) { | ||
| return state.Actor.GetStatus() == ateapipb.Actor_STATUS_RUNNING, nil | ||
| } | ||
| func (s *CallAteletRestoreStep) CheckPrerequisite(ctx context.Context, input *ResumeInput, state *ResumeState) error { |
There was a problem hiding this comment.
I added more checks in CheckPrerequisite to make sure that, if we retry ResumeActor, the Worker assignment and eligibility are still correct for the actor.
…o check the worker eligibility again if previous attempt failed after setting actor to RESUMING state. If the worker becomes ineligible, set the actor to crashed state.
Zoe Zhao (zoez7)
force-pushed
the
check-prerequisite
branch
from
5ba9448 to
6d976cf
Compare
3 participants
Fixes #369
Add a CheckPrerequisite method to the WorkflowStep interface, called by RunWorkflow after IsComplete returns false and before Execute, so that each workflow validates its actor state-machine edge up front while retried (reentrant) workflows still fast-forward past completed steps.