Where do the task_state_handlers get called? In the main flow (scheduler) process or on the workers (dask, subprocess etc.) I'm passing an object to task_state_handlers through a closure, and with sequential/threads LocalDaskExecutor they work OK, but with processes it's clear that the passed object isn't the same instance for all task runs I was under the impression that the scheduler catches state changes and calls all the callbacks, while the executor is only responsible for, well, the execution of work.

They get called on the dask workers. The executor is responsible for submitting the individual Runner pipelines for work; Runner pipelines include such things are managing state transitions for the individual task that was submitted.

Thanks for the clarification! So task states are not getting reported to the original process? I'm trying to buffer messages going from prefect to my storage backend and this buffer only makes sense if it's relatively long-lived (minutes and hours, not seconds).

final task states are always reported back to the original process, but intermediate ones are not necessarily

@Chris White Is there a recommended way of passing these intermediate states back to the flow process. Say if we wanted to take an action on the flow based on if an exception occurred in a task or not.

Hi @J Grover - for taking action based on an exception within a task, you could use a `state_handler`; for example, to do something custom if a

TypeError

is raised within a task:

def state_handler(tt, old_state, new_state):
    if new_state.is_failed() and isinstance(new_state.result, TypeError):
        # do something custom
    return new_state

if you really needed to talk back to the Flow’s process for some reason, I would recommend directly utilizing dask’s

run_on_scheduler

method: https://distributed.dask.org/en/latest/api.html#distributed.Client.run_on_scheduler

I am more after the latter. So there isn't a native Prefect way of propagating such task state changes to the main process while the flow is running?

Honestly we’ve never even considered it - could you explain your use case for needing this?

Well in my case, flow and it's tasks are reporting their state changes to a server via API. API client wrapper is buffered: I add messages to a deque and then try to send them, so that they are kept there in case the webserver is down. If the flow is stopped, I persist the deque to a file that will be loaded on the next flow.run() call, so theoretically no state will be lost and eventually all messages will reach the server. Because communication with the server is more complex than just calling

<http://requests.post|requests.post>()

, I'd like to handle it all in one place. The buffering only make sense if the number of buffers is relatively low. Right now, it's 1 queue for flow + 1 attached to each worker. This works, but I'm unable to persist buffered state changes across worker restarts.

You could put your API payloads into a distributed Queue (https://distributed.dask.org/en/latest/api.html#distributed.Queue), and then read off the queue in a background thread in the flow’s process

This is super cool, thank you for directing me there. I thought that an external queue would be a solution, but didn't want to deploy a standalone mq for this.