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

[fjetter]

which causes the worker to automatically reconnect to the scheduler

We removed the worker reconnect in #6361 🤔

[crusaderky]

worker-connect-

which causes the worker to automatically reconnect to the scheduler

We removed the worker reconnect in #6361 thinking

Well something is clearly calling Worker._register_with_scheduler after Scheduler.handle_stream crashes, calling await comm.close() in its finally clause.

[crusaderky]

Whoops. I just added a test for the edge case where a task raises Reschedule(), but the client has already released the future... and the worker state machine just horribly fell apart 🥳

  [Previous line repeated 2941 more times]
  File "/home/crusaderky/github/distributed/distributed/worker_state_machine.py", line 2299, in _transition
    recs, instructions = self._transition(
  File "/home/crusaderky/github/distributed/distributed/worker_state_machine.py", line 2279, in _transition
    func = self._TRANSITIONS_TABLE.get((start, finish))
RecursionError: maximum recursion depth exceeded in comparison

[crusaderky]

worker-connect-

which causes the worker to automatically reconnect to the scheduler

We removed the worker reconnect in #6361 thinking

Well something is clearly calling Worker._register_with_scheduler after Scheduler.handle_stream crashes, calling await comm.close() in its finally clause.

Mistery solved: the original worker died and was not replaced. The finally clause of Scheduler.handle_worker calls remove_worker, passing the original stimulus_id - which is potentially hours old and was not used anywhere before.
Remediation to follow.