future.retry() retries futures regardless of its status

[Mirandatz]

What happened:
I was checking my logs and noticed that some futures were executed more than once.
It appears that future.retry() will unconditionally re-schedule* the future.

• I'm not sure if re-scheduling is the proper expression.

What you expected to happen:
I expected that future.retry() would re-scheduled ONLY futures that failed,
i.e.: that had a status == "lost" or "error".
But it appears that it re-schedules futures even if they are 'pending'
That is what I understood from the documentation:

https://distributed.dask.org/en/latest/api.html#distributed.Client.retry

retry(futures, asynchronous=None)[source]
Retry failed futures

Minimal Complete Verifiable Example:
In the code below, I expected to ready "started" once; when the future was started.
But it is printed infinitely many times and the task never terminates.

import time

import distributed


def ughz():
    print("started!")
    time.sleep(2)


def main():
    cluster = distributed.LocalCluster(threads_per_worker=1, n_workers=1)
    client = distributed.Client(cluster)
    f = client.submit(func=ughz, pure=False)
    while f.status != "finished":
        print(f.status)
        print()
        f.retry()


if __name__ == '__main__':
    main()

Anything else we need to know?:
It was hard to find which values a future.status can assume.
I had to check client.py source code.

Environment:

• Dask version: 2020.12.0
• Python version: 3.8.5
• Operating System: CentOS Linux release 7.2.1511 (Core)
• Install method (conda, pip, source): conda

[mazen-isstech]

My gripe with Future.retry() is that it changes the status from "finished" to "pending" to "lost", and back to "finished". So at some point the task is considered lost for some reason, even though the task is doing well.