Workflow purge#6163
Conversation
Signed-off-by: Ryan Lettieri <[email protected]>
Signed-off-by: Ryan Lettieri <[email protected]>
Signed-off-by: Ryan Lettieri <[email protected]>
Signed-off-by: Ryan Lettieri <[email protected]>
Signed-off-by: Ryan Lettieri <[email protected]>
| } | ||
|
|
||
| var resp *invokev1.InvokeMethodResponse |
There was a problem hiding this comment.
| } | |
| var resp *invokev1.InvokeMethodResponse | |
| } | |
| var resp *invokev1.InvokeMethodResponse |
Because there'll be a change on this file getting merged very soon and I want to spare a merge conflict :)
|
|
||
| // Try to load activity state. If we find any, that means the activity invocation is a duplicate. | ||
| if state, err := a.loadActivityState(ctx, actorID, ar.Generation); err != nil { | ||
| var state activityState |
There was a problem hiding this comment.
Uhm something feels odd here. state is not read from a.loadActivityState anymore, so it will always be empty? and if state.Generation> 0 will always be false? I'm not sure what this should have been.
| if methodName == "PurgeWorkflowState" { | ||
| if err := a.purgeActivityState(ctx, actorID, state); err != nil { | ||
| return nil, err | ||
| } |
There was a problem hiding this comment.
Should this method return after processing this? We don't need to schedule a reminder, do we?
|
|
||
| func getActivityInvocationKey(generation uint64) string { | ||
| return fmt.Sprintf("activityreq-%d", generation) | ||
| // return fmt.Sprintf("activityreq-%d", generation) |
There was a problem hiding this comment.
Why removing the generation?
|
|
||
| "github.com/dapr/components-contrib/workflows" | ||
| componentsV1alpha1 "github.com/dapr/dapr/pkg/apis/components/v1alpha1" | ||
| componentsV1alpha1 "github.com/dapr/dapr/pkg/apis/components/v1alpha1" // This will be removed |
There was a problem hiding this comment.
Forgot to remove this and ComponentDefinition? :)
|
|
||
| func (c *workflowEngineComponent) Purge(ctx context.Context, req *workflows.PurgeRequest) error { | ||
| if req.InstanceID == "" { | ||
| return fmt.Errorf("a workflow instance ID is required") |
There was a problem hiding this comment.
| return fmt.Errorf("a workflow instance ID is required") | |
| return errors.New("a workflow instance ID is required") |
If no format is included, please use errors.New (it's faster and makes less allocs)
|
|
||
| // Loop through the history of the loaded workflow state and schedule purge calls for all the activities that have been completed | ||
|
|
||
| // RRL TODO: Looping through history and invoking each actor once. They are not in the same transaction. Eqch transaction happens as its own call |
There was a problem hiding this comment.
As per the initial review during our 1:1 earlier today, this needs to change - although I admit I don't fully understand how this works so I can't tell you exactly how to fix this yet 🙃
However, the problem is that the "purge" needs to happen in an atomic way. We discussed this and you explained that you need to delete 3 things:
- Each activity (this can be more than one)
- The history object
- The inbox object
These all have to be removed at once. We do that in Dapr by using the transactional state stores and performing a "transaction" (which are just operations performed atomically as a batch), but here, each activity is removed separately, in its own "transaction" (with a single opeation). Each activity is removed by the actor that you're invoking within the loop, and an error in the loop causes the entire method to exit, which means that some state will have been removed, but not all.
There was a problem hiding this comment.
Hi @ItalyPaleAle. The state exists across potentially multiple actors, and AFAIK, we don't support transactions across multiple actors. We instead go for the next best thing, which is to do independent purge operations in a particular order so that failures that happen midway can be retried until there is finally a successful completion. This is why, for example, we delete state for each activity first before deleting the history (without the history, we wouldn't know which actors need to be deleted). Does that make sense? Let me know if you feel we need to meet and discuss this more in depth.
There was a problem hiding this comment.
I see. But what happens if there's a crash while the process is not complete?
There was a problem hiding this comment.
The crash will leave you in a partially deleted state. However, the workflow will still be visible from the GetWorkflow management APIs, and re-issuing the purge call should succeed.
Signed-off-by: Ryan Lettieri <[email protected]>
| rpc RaiseEventWorkflowAlpha1 (RaiseEventWorkflowRequest) returns (WorkflowActivityResponse) {} | ||
|
|
||
| // Purge Workflow | ||
| rpc PurgeWorkflowAlpha1 (PurgeWorkflowRequest) returns (WorkflowActivityResponse) {} |
There was a problem hiding this comment.
Standard practice for APIs that don't return anything is to use google.protobuf.Empty. I suggest we do the same here.
There was a problem hiding this comment.
Looks like this got missed?
| } | ||
|
|
||
| func (a *UniversalAPI) RaiseEventWorkflowAlpha1(ctx context.Context, in *runtimev1pb.RaiseEventWorkflowRequest) (*runtimev1pb.RaiseEventWorkflowResponse, error) { | ||
| func (a *UniversalAPI) RaiseEventWorkflowAlpha1(ctx context.Context, in *runtimev1pb.RaiseEventWorkflowRequest) (*runtimev1pb.WorkflowActivityResponse, error) { |
There was a problem hiding this comment.
I suggest reverting these changes because I'm making additional changes to these signatures in a nother PR.
| for _, e := range state.History { | ||
| if ts := e.GetTaskScheduled(); ts != nil { | ||
| targetActorID := getActivityActorID(actorID, e.EventId) | ||
| eventData, err := backend.MarshalHistoryEvent(e) |
There was a problem hiding this comment.
We shouldn't be including the event data in the purge request.
There was a problem hiding this comment.
This still needs to be addressed (though the problem has moved to a new line now)
Signed-off-by: Ryan Lettieri <[email protected]>
Signed-off-by: Ryan Lettieri <[email protected]>
Signed-off-by: Ryan Lettieri <[email protected]>
| req := invokev1. | ||
| NewInvokeMethodRequest(PurgeWorkflowStateMethod). | ||
| WithActor(be.config.workflowActorType, string(id)). | ||
| WithContentType(invokev1.OctetStreamContentType) |
There was a problem hiding this comment.
You can remove the WithContentType line since we're not sending any content.
| rpc RaiseEventWorkflowAlpha1 (RaiseEventWorkflowRequest) returns (WorkflowActivityResponse) {} | ||
|
|
||
| // Purge Workflow | ||
| rpc PurgeWorkflowAlpha1 (PurgeWorkflowRequest) returns (WorkflowActivityResponse) {} |
There was a problem hiding this comment.
Looks like this got missed?
| for _, opt := range GetTestOptions() { | ||
| t.Run(opt(engine), func(t *testing.T) { | ||
| id, err := client.ScheduleNewOrchestration(ctx, "ActivityChainToPurge") | ||
| if assert.NoError(t, err) { |
There was a problem hiding this comment.
To reduce nesting, let's replace assert.NoError throughout this test with require.NoError if remove the if-blocks
| for _, e := range state.History { | ||
| if ts := e.GetTaskScheduled(); ts != nil { | ||
| targetActorID := getActivityActorID(actorID, e.EventId) | ||
| eventData, err := backend.MarshalHistoryEvent(e) |
There was a problem hiding this comment.
This still needs to be addressed (though the problem has moved to a new line now)
| } | ||
| }) | ||
| } | ||
| } |
There was a problem hiding this comment.
Let's add one more test in order to ensure we're covering the multi-generation case. Basically, copy the TestContinueAsNewWorkflow logic and do a purge against that workflow state, validating that all state store content was successfully removed.
Signed-off-by: Ryan Lettieri <[email protected]>
Signed-off-by: Ryan Lettieri <[email protected]>
Signed-off-by: Ryan Lettieri <[email protected]>
Signed-off-by: Ryan Lettieri <[email protected]>
Signed-off-by: Ryan Lettieri <[email protected]>
| cachedState := result.(activityState) | ||
|
|
||
| // Make sure the cached state is for the same generation of the workflow. | ||
| if cachedState.Generation == generation { |
There was a problem hiding this comment.
Now that we're not using the Generation field, let's delete it from activityState.
| } | ||
|
|
||
| for _, item := range keysPostPurge { | ||
| if strings.Contains(item, string(id)) { |
There was a problem hiding this comment.
Should we use assert.True(t, strings.Contains(item, string(id))) here?
| client, engine, stateStore := startEngineAndGetStore(ctx, r) | ||
| for _, opt := range GetTestOptions() { | ||
| t.Run(opt(engine), func(t *testing.T) { | ||
| // Second Test |
There was a problem hiding this comment.
Is this comment still relevant or can it be removed?
|
|
||
| // Loop through the history of the loaded workflow state and schedule purge calls for all the activities that have been completed | ||
|
|
||
| // RRL TODO: Looping through history and invoking each actor once. They are not in the same transaction. Each transaction happens as its own call |
There was a problem hiding this comment.
This doesn't seem like a TODO, but more like a description of the code.
| // Loop through the history of the loaded workflow state and schedule purge calls for all the activities that have been completed | ||
|
|
||
| // RRL TODO: Looping through history and invoking each actor once. They are not in the same transaction. Each transaction happens as its own call | ||
| // We are deleting one by one in sequence and not in builk. |
There was a problem hiding this comment.
This comment seems redundant with the previous one (and has a typo). I suggest we clean up these comments.
| req := invokev1. | ||
| NewInvokeMethodRequest(PurgeWorkflowStateMethod). | ||
| WithActor(wf.config.activityActorType, targetActorID). | ||
| WithRawDataBytes(activityRequestBytes). |
There was a problem hiding this comment.
Now that we don't use the generation anymore, can we remove the payload completely from the actor invocation?
|
|
||
| // The logic/for loop below purges/removes any leftover state from a completed or failed activity | ||
| // TODO: for optimization make multiple go routines and run them in parallel | ||
| for _, e := range state.Inbox { |
There was a problem hiding this comment.
Let's move this logic into its own method.
| func addPurgeStateOperations(req *actors.TransactionalRequest, keyPrefix string, events []*backend.HistoryEvent) error { | ||
| // TODO: Investigate whether Dapr state stores put limits on batch sizes. It seems some storage | ||
| // providers have limits and we need to know if that impacts this algorithm: | ||
| // https://learn.microsoft.com/azure/cosmos-db/nosql/transactional-batch#limitations |
There was a problem hiding this comment.
I think we should open a bug to track doing purges in smaller batches to avoid running into transactional batch limitations.
Signed-off-by: Ryan Lettieri <[email protected]>
Codecov Report
@@ Coverage Diff @@
## master #6163 +/- ##
==========================================
- Coverage 64.96% 64.76% -0.21%
==========================================
Files 189 189
Lines 18335 18471 +136
==========================================
+ Hits 11912 11962 +50
- Misses 5492 5565 +73
- Partials 931 944 +13
|
cgillum
left a comment
cgillum
left a comment
There was a problem hiding this comment.
Note to other reviewers that we decided to remove the de-dupe detection for activities since it was turning out to be a problematic (undocumented) optimization. We'll revisit whether it makes sense to add it back in the future.
A couple more things before signing off.
Signed-off-by: Ryan Lettieri <[email protected]>
|
Once this PR is merged, @RyanLettieri you'll need to generate the protos in the various SDK PRs you have open for workflow management. Implement the purge operation too, please. |
Signed-off-by: Bernd Verst <[email protected]>
5 participants
Description
Addition of Purge method for dapr workflow.
Issue reference
Note that this PR will not pass until the following PR is submitted into components-contrib: dapr/components-contrib#2729
Please reference the issue this PR will close: #[6043]
Checklist
Please make sure you've completed the relevant tasks for this PR, out of the following list: