Interrupt heartbeating activity on pause #854
Conversation
temporalio/client.py
Outdated
| ) | ||
| if resp_by_id.cancel_requested: | ||
| raise AsyncActivityCancelledError() | ||
| if resp_by_id.activity_paused: |
There was a problem hiding this comment.
Hrmm, looks like we never added "raises" docs to AsyncActivityHandle.heartbeat, wonder if we should now
THardy98
force-pushed
the
support_activity_pause
branch
from
d5107b0 to
9763505
Compare
THardy98
force-pushed
the
support_activity_pause
branch
2 times, most recently
from
5c8299a to
4451e77
Compare
temporalio/testing/_activity.py
Outdated
| if cancellation_details: | ||
| self.cancellation_details.set_details(cancellation_details) |
There was a problem hiding this comment.
I think this needs to always be set, even if not provided. May want to change param to be non-option w/ a default of temporalio.activity.ActivityCancellationDetails(). Usually using a default param of a full expression is bad in Python because it isn't called for each call, but since the details are immutable, sharing the default for everyone is safe.
There was a problem hiding this comment.
I think this needs to always be set, even if not provided.
Why?
May want to change param to be non-option w/ a default of temporalio.activity.ActivityCancellationDetails()
Sort of defeats the purpose of being able to return None without checking all fields are False, in which case, maybe the holder indirection isn't so useful.
There was a problem hiding this comment.
I think the logic needs to be "you can guarantee activity.cancellation_details() is not None when the activity is being canceled", but this breaks that guarantee.
There was a problem hiding this comment.
Unresolving, I think this is still an issue. We should always set an instance of this on cancel in test environment. So the parameter should default to an empty instance with no optional accepted.
tests/worker/test_workflow.py
Outdated
| except (CancelledError, asyncio.CancelledError): | ||
| return activity.cancellation_details() |
There was a problem hiding this comment.
Need some tests to confirm what happens when this is not caught
There was a problem hiding this comment.
Added a test for pausing/unpausing activities, we don't catch cancel errors.
Added the unpause activity method to the bridge client - it was missing.
I made it such that a cancelled error on activity pause returns an activity task failure w/ application failure:
- When going through the default failure converter, a canceled error populates cancel failure info, but core expects application failure info because we are populating the
failedfield of the completion instead ofcancel. - A bit of a weird situation, because we can't populate cancel for the reason you mentioned above (we tell server to cancel and server complains that a cancel hasn't been requested yet - like you mention above)
I can revert this change if you prefer, I don't think it causes any notable server effect anyway (state of paused activities are suspended server-side, iiuc), but probably a good thing to have failures sent properly just as a general rule
THardy98
force-pushed
the
support_activity_pause
branch
2 times, most recently
from
c5ca36e to
c3634fe
Compare
THardy98
force-pushed
the
support_activity_pause
branch
2 times, most recently
from
4977651 to
e055766
Compare
THardy98
force-pushed
the
support_activity_pause
branch
from
8257841 to
ab65714
Compare
temporalio/client.py
Outdated
| ) | ||
| if resp_by_id.cancel_requested: | ||
| if resp_by_id.cancel_requested or resp_by_id.activity_paused: | ||
| raise AsyncActivityCancelledError() |
There was a problem hiding this comment.
We can do this in a separate issue if you want, but I think we need to put cancellation details on this exception now that there is more than one reason why it may get canceled.
tests/worker/test_workflow.py
Outdated
| # Wait for next heartbeat to propagate the cancellation. Unpausing before the heartbeat | ||
| # will show activity as unpaused to core. Consequently, it will *not* issue an activity cancel. | ||
| time.sleep(0.3) |
There was a problem hiding this comment.
Is there a way to get rid of potentially flaky sleeps and wait for situations that are needed? For instance in temporalio/sdk-dotnet#482 I used the presence of the final heartbeat details to confirm activity completed (kinda).
There was a problem hiding this comment.
replaced with a helper that waits for the next heartbeat to elapse
…ellation errors from external activities
THardy98
force-pushed
the
support_activity_pause
branch
from
7977621 to
56913c4
Compare
tests/helpers/__init__.py
Outdated
| raise AssertionError( | ||
| f"Activity with ID {activity_id} not found in pending activities" | ||
| ) | ||
| if current_info.last_heartbeat_time == initial_heartbeat_time: |
There was a problem hiding this comment.
What we did in .NET was just check the final heartbeat details (i.e. we call heartbeat in a finally with certain details). I wonder if this is technically racy because what if the final heartbeat happened and the activity returned before wait_for_next_heartbeat_cycle was called? Also can move this logic to inside the test it's used in since it's not a general helper.
There was a problem hiding this comment.
I think the race condition you propose is possible, in which case, we'd be stuck on the assertion because the activity wouldn't emit a subsequent heartbeat (the activity is paused so we wouldn't get a heartbeat from a retry).
I've changed it to just check if heartbeat details exists (not the exact details payload). This is enough to verify that the activity has indeed finished executing and is ready to be unpaused.
|
thanks for the reviews :) |
3 participants
What was changed
Adds support to interrupt heartbeating activities from pause requests.
Why?
Functionally support users pausing their activities via the CLI
Closes Heartbeating activities should be interrupted when the activities are paused. #812
How was this tested:
Simple integration test
Any docs updates needed?
Unsure