What happened: When an Actor raises an exception within an async method, its result hangs indefinitely.

What you expected to happen: The exception to be propagated to the Future resulting from the method being called.

Minimal Complete Verifiable Example:

import dask.distributed as d
import time

class Actor:
    async def something(self):
        raise ValueError('beep')


def main():
    c = d.Client()
    a = c.submit(Actor, actor=True).result()

    p = a.something()
    time.sleep(0.1)
    print(f'About to resolve')
    print(p.result())

if __name__ == '__main__':
    main()

Anything else we need to know?:

Environment:

• Dask version: 2021.04.1
• Python version: 3.8.1
• Operating System: Ubuntu
• Install method (conda, pip, source): pip