Updates PrefectDistributedFuture to use an event websocket to determine task completion

[desertaxle]

Example

Flows like this still work (assuming you're running the appropriate task worker):

from prefect.task_runners import PrefectTaskRunner
from prefect import flow

from tasks import square, neg, summation


@flow(task_runner=PrefectTaskRunner())
def example_flow():
    A = square.map(range(10))
    B = neg.map(A)
    total = summation.submit(B)

    return total.result()


if __name__ == "__main__":
    result = example_flow()
    print(result)

Checklist

• This pull request references any related issue by including "closes <link to issue>"
  • If no issue exists and your change is not a small fix, please create an issue first.
• If this pull request adds new functionality, it includes unit tests that cover the changes
• This pull request includes a label categorizing the change e.g. maintenance, fix, feature, enhancement, docs.

For documentation changes:

• This pull request includes redirect settings in mint.json for files that are removed or renamed.

For new functions or classes in the Python SDK:

• This pull request includes helpful docstrings.
• If a new Python file was added, this pull request contains a stub page in the Python SDK docs and an entry in docs/mint.json navigation.

[zzstoatzz]

this is really cool! I really like the idea of using the consumer here

wondering if you ran into any pubsub "backup" that caused us to wait longer than we really should have for the event to trickle through? just thinking through the tradeoffs

[zzstoatzz]

🙇

[chrisguidry]

Looks good to me!

[desertaxle]

wondering if you ran into any pubsub "backup" that caused us to wait longer than we really should have for the event to trickle through? just thinking through the tradeoffs

Our latency has improved a ton since we removed logs from the rest of the events stream, but it is something that we'll need to keep an eye on.