…-task-engine

…-task-engine

…-task-engine

Co-authored-by: Nathan Nowack <thrast36@gmail.com>

…11737)

Co-authored-by: Chris Guidry <chris.g@prefect.io>

… into init-task-engine

…refectHQ/prefect into init-task-engine

In earlier work, we've introduced autonomous task scheduling, where tasks
outside a flow run are created as scheduled and picked up by one or more
processes running `Task.serve`.  In our initial implementation, we used a
polling approach where each `TaskSever` would make requests from the API to
look for any tasks that were currently `Scheduled`, and then move them to
`Running` as they entered the task engine.

This work introduces a new mechanism for `TaskServer`s to get work from their
Prefect Server: a long-lived websocket connection subscribed to a queue of
`TaskRun`s to be worked.  Because the Prefect Server is a singleton, it can
govern a queue in-memory that will be distributed out among each of the
`TaskServer`s to make a simple task brokering system.

The websocket implementation is modeled on the `events/in` and `events/out`
websockets in Prefect Cloud, and it's expected that we'd negotiate
authentication in a common way across all websockets.

Note: this does not address issues of resiliency, like what happens if the
Prefect Server is restarted (in-flight tasks would be lost), or if there are
no `TaskServer`s draining the Queue (the Prefect Server would eventually run out
of memory), or if a `TaskServer` died before transitioning a task to `Running`
(the task would remain `Scheduled` and never get picked up).  These are some
of the items I'd like to address in future work if we like this direction.

Note: there are no tests for this new subsystem yet.