Scheduler.reschedule() does not work

[crusaderky]

Scheduler.reschedule() is called:

1. from Worker.transition_long_running_rescheduled
2. from Worker.transition_executing_rescheduled
3. from stealing.py
4. directly from a bunch of unit tests

If you add "assert False" at the top of the method, only the tests that invoke it directly fail.
Notably, among them there's test_reschedule - in other words, there is nothing at the moment that tests that worker transitions and work stealing are successfully using the method.

[crusaderky]

Correction: there is one test that triggers rescheduling from the worker, test_worker.py::test_reschedule.

At the moment of writing, the call to reschedule() from the Worker does NOT work due to a mismatched function signature. A TypeError can be spotted in the log by eagle-eyed users.
Once #6339 fixed the call, the worker state machine broke:

Traceback (most recent call last):
  File "/home/crusaderky/github/distributed/distributed/worker.py", line 4277, in validate_task
    self.validate_task_cancelled(ts)
  File "/home/crusaderky/github/distributed/distributed/worker.py", line 4245, in validate_task_cancelled
    assert ts._next is None  # We'll always transition to released after it is done
AssertionError

[crusaderky]

CC @fjetter

[crusaderky]

The reason why, when you raise Reschedule() in a task, the scheduler effectively reschedules the task is that:

1. the worker sends the scheduler a reschedule message, which doesn't match the signature of Scheduler.reschedule
2. which causes Scheduler.handle_stream to raise TypeError
3. which causes the batched stream channel to collapse
4. which causes the worker to automatically reconnect to the scheduler
5. the connection event causes the scheduler to transition the task processing->released->waiting->processing

Scheduler log (note the stimulus ids):

- ('x1', 'released', 'waiting', {'x1': 'processing'}, 'update-graph-1656417887.9150004', 1656417887.9152453)
- ('x1', 'waiting', 'processing', {}, 'update-graph-1656417887.9150004', 1656417887.9161289)
- ('x1', 'processing', 'released', {'x1': 'waiting'}, 'worker-connect-1656417887.7426143', 1656417888.023502)
- ('x1', 'released', 'waiting', {'x1': 'processing'}, 'worker-connect-1656417887.7426143', 1656417888.0235095)
- ('x1', 'waiting', 'processing', {}, 'worker-connect-1656417887.7426143', 1656417888.0235472)
- ('x1', 'processing', 'memory', {}, 'task-finished-1656417888.223756', 1656417888.2246292)

... 🤔 ...

I mean, if it works, don't touch it, amirite? 😆 😆 😆 😆 😆

[crusaderky]

Once #6339 fixed the call, the worker state machine broke:

Traceback (most recent call last):
  File "/home/crusaderky/github/distributed/distributed/worker.py", line 4277, in validate_task
    self.validate_task_cancelled(ts)
  File "/home/crusaderky/github/distributed/distributed/worker.py", line 4245, in validate_task_cancelled
    assert ts._next is None  # We'll always transition to released after it is done
AssertionError

I can no longer reproduce this