except (asyncio.exceptions.CancelledError, prefect.exceptions.TerminationSignal):
        # if the flow is cancelled, we should just do nothing here
        # the caller is responsible for handling the state sync with camote and tortilla
        # this is because the flow could be cancelled without even starting
        log.warn(f"flow {task_id} is cancelled, terminate the processes")
        return Cancelled(message=f"Cancelled: {task_id} is cancelled")

flow

on_crashed=[on_failure_noitiy_slack],
    on_cancellation=[_on_cancelled],
    on_completion=[_on_completion],

def _on_cancelled(flow, flow_run, state):
    print(f"flow {flow_run.id} is {state}, deregister the service address")
    print(f"flow run meta: {flow.dict()} {flow_run.dict()}")
    print(f"flow run context: {prefect.context.get_run_context().dict()}")

on_cancellation