AssertionError in Scheduler.restart: assert not self.tasks

[crusaderky]

https://github.com/coiled/coiled-runtime/actions/runs/3686850652/jobs/6239652435

What happened:

1. test_spill failed for whatever reason. When it failed, the workers were full of tasks and their event loop was severely slowed down by spilling.
2. The teardown of test_spill garbage-collected the Client. It is unknown if the information had enough time to reach the Scheduler.
3. Immediately afterwards, the setup of test_tensordot_stress created a new Client and called client.restart()
4. For some reason, Scheduler.client_releases_keys failed to transition all tasks to forgotten; Scheduler.restart consequently failed on line 5730:
   

   distributed/distributed/scheduler.py

   Lines 5722 to 5730 in 7fb9c48

   	for cs in self.clients.values():
   	self.client_releases_keys(
   	keys=[ts.key for ts in cs.wants_what],
   	client=cs.client_key,
   	stimulus_id=stimulus_id,
   	)
   	self._clear_task_state()
   	assert not self.tasks

There is no cluster dump of test_tensordot_stress.
cluster dump of test_spill (taken between step 1 and 2 above): s3://coiled-runtime-ci/test-scratch/cluster_dumps/spill-9ad15a09/stability.test_spill.py.test_spilling[False].msgpack.gz

CC @fjetter

[gjoseph92]

According to @fjetter I believe this is related to #7396?

[gjoseph92]

test_forget_tasks_while_processing added in #7402 fails on main:

            await c.close()
            del c, first_batch
    
>           await async_wait_for(lambda: len(s.tasks) == len(second_batch), 5)

distributed/tests/test_scheduler.py:413: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

predicate = <function test_queued_release_multiple_workers.<locals>.<lambda> at 0x7f9a50098940>, timeout = 5, fail_func = None, period = 0.05

    async def async_wait_for(predicate, timeout, fail_func=None, period=0.05):
        deadline = time() + timeout
        while not predicate():
            await asyncio.sleep(period)
            if time() > deadline:
                if fail_func is not None:
                    fail_func()
>               pytest.fail(f"condition not reached until {timeout} seconds")
E               Failed: condition not reached until 5 seconds

This failure—tasks are not removed from the scheduler after their client disconnects—seems like the same thing as this issue. That makes me confident that #7402 would fix it, and that the test I added is a reproducer.

[fjetter]

According to @fjetter I believe this is related to #7396?

Yes