SpecCluster correct state can cause inconsistencies

[fjetter]

There are a bunch of problems in SpecCluster._correct_state_internal such that I believe it should be rewritten

• If one worker fails during startup, all workers are rejected. This can cause the cluster to spin up too many workers
• Similarly, while closing, if one worker fails but others properly shut down, they are not removed from the internal state
• correct_state_internal is only called once, i.e. any kind of exception would abort the entire up/downscaling without further attempt to correct the state.
• If the cluster is closing while correct_state is running, nothing is actually cancelled.
• self._correct_state_waiting is actually never cancelled.
• SpecCluster.scale schedules a callback to self _correct_state. This can cause all sorts of race conditions, e.g. by creating more futures even if the cluster is already closing

This list probably continues long and most issues could be addressed individually but I believe we're better off rewriting this section.

cc @graingert

[graingert]

If one worker fails during startup, all workers are rejected. This can cause the cluster to spin up too many workers

it's actually sort of worse - the workers get 2 chances to startup, one (concurrently) in asyncio.wait(workers) and another (sequentially) in await w # for tornado gen.coroutine support

distributed/distributed/deploy/spec.py

Lines 349 to 352 in e1e4385

	await asyncio.wait(workers)
	for w in workers:
	w._cluster = weakref.ref(self)
	await w # for tornado gen.coroutine support

I think this is supposed to be more like

tasks = [asyncio.create_task(worker) for worker in workers]
await asyncio.wait(tasks)
for t in tasks:
    w = await t
    w._cluster = weakref.ref(self)

[fjetter]

Xref #4606