async def METHOD_NAME(c, s, a, b):
    lock = distributed.Lock()
    await lock.acquire()
    async def wait_and_raise(*args, **kwargs):
        async with lock:
            raise RuntimeError()
    f = c.submit(wait_and_raise, key="cancel", workers=[a.address])
    p = MultiProgress([f], scheduler=s, complete=True)
    await p.setup()
    await wait_for_state(f.key, "executing", a)
    f.release()
    await wait_for_state(f.key, "cancelled", a)
    assert p.status == "error"
    assert p.all_keys.keys() == {"cancel"}