@@ -3487,6 +3487,72 @@ async def test_workflow_local_activity_backoff(client: Client):
34873487 )
34883488
34893489
3490+ @activity .defn
3491+ async def retry_later_activity () -> None :
3492+ # A delay above the default 60s local_retry_threshold makes Core back off
3493+ # with a workflow timer instead of retrying in memory
3494+ raise ApplicationError ("Retry later" , next_retry_delay = timedelta (seconds = 90 ))
3495+
3496+
3497+ @workflow .defn
3498+ class LocalActivityBackoffCancelWorkflow :
3499+ def __init__ (self ) -> None :
3500+ self ._cancel_activity = False
3501+
3502+ @workflow .run
3503+ async def run (self , cancel_from_workflow : bool ) -> str :
3504+ handle = workflow .start_local_activity (
3505+ retry_later_activity , start_to_close_timeout = timedelta (seconds = 10 )
3506+ )
3507+ if cancel_from_workflow :
3508+ await workflow .wait_condition (lambda : self ._cancel_activity )
3509+ handle .cancel ()
3510+ try :
3511+ await handle
3512+ except asyncio .CancelledError :
3513+ return "activity cancelled"
3514+ await handle
3515+ return "activity completed"
3516+
3517+ @workflow .signal
3518+ def cancel_activity (self ) -> None :
3519+ self ._cancel_activity = True
3520+
3521+
3522+ @pytest .mark .parametrize ("cancel_from_workflow" , [False , True ])
3523+ async def test_workflow_local_activity_cancel_during_backoff (
3524+ client : Client , cancel_from_workflow : bool
3525+ ):
3526+ async with new_worker (
3527+ client , LocalActivityBackoffCancelWorkflow , activities = [retry_later_activity ]
3528+ ) as worker :
3529+ handle = await client .start_workflow (
3530+ LocalActivityBackoffCancelWorkflow .run ,
3531+ cancel_from_workflow ,
3532+ id = f"workflow-{ uuid .uuid4 ()} " ,
3533+ task_queue = worker .task_queue ,
3534+ )
3535+
3536+ async def backoff_timer_started () -> None :
3537+ async for e in handle .fetch_history_events ():
3538+ if e .event_type == EventType .EVENT_TYPE_TIMER_STARTED :
3539+ return
3540+ assert False , "Backoff timer not started"
3541+
3542+ await assert_eventually (backoff_timer_started )
3543+ if cancel_from_workflow :
3544+ await handle .signal (LocalActivityBackoffCancelWorkflow .cancel_activity )
3545+ assert "activity cancelled" == await handle .result ()
3546+ else :
3547+ await handle .cancel ()
3548+ with pytest .raises (WorkflowFailureError ) as err :
3549+ await handle .result ()
3550+ assert isinstance (err .value .cause , CancelledError )
3551+ event_types = [e .event_type async for e in handle .fetch_history_events ()]
3552+ assert EventType .EVENT_TYPE_TIMER_CANCELED in event_types
3553+ assert EventType .EVENT_TYPE_WORKFLOW_TASK_FAILED not in event_types
3554+
3555+
34903556deadlock_thread_event = threading .Event ()
34913557
34923558
0 commit comments