`distributed.client.FutureCancelledError` when computing a `da.from_delayed(future)` when `future` goes out of scope.

[TomAugspurger]

Describe the issue:

This seems similar to #9041, which was fixed by dask/dask#11886, but for arrays. It started failing with dask/dask@0194ef4 / dask/dask#11881.

Minimal Complete Verifiable Example:

from distributed import Client
import dask.array as da
import numpy as np


def main():
    arr = np.zeros((10, 10))
    meta = arr[:0, :0]

    with Client(n_workers=2) as client:
        fut = client.scatter(arr)
        result = da.from_delayed(fut, shape=arr.shape, meta=meta)
        result.compute()
        print("Compute 1, done")

        del fut

        result.compute()
        print("Compute 2, done")


if __name__ == "__main__":
    main()

The first .compute() succeeds. The second .compute(), after the del fut, fails with

❯ python debug.py
Compute 1, done
Traceback (most recent call last):
  File "/home/nfs/toaugspurger/gh/rapidsai/dask-upstream-testing/debug.py", line 44, in <module>
    main()
  File "/home/nfs/toaugspurger/gh/rapidsai/dask-upstream-testing/debug.py", line 21, in main
    result.compute()
  File "/home/nfs/toaugspurger/gh/rapidsai/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/base.py", line 373, in compute
    (result,) = compute(self, traverse=False, **kwargs)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/nfs/toaugspurger/gh/rapidsai/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/base.py", line 681, in compute
    results = schedule(expr, keys, **kwargs)
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/nfs/toaugspurger/gh/rapidsai/dask-upstream-testing/.venv/lib/python3.12/site-packages/distributed/client.py", line 2400, in _gather
    raise exception.with_traceback(traceback)
distributed.client.FutureCancelledError: finalize-hlgfinalizecompute-199c06cde2ba4625a856dd08fc888898 cancelled for reason: lost dependencies.

Anything else we need to know?:

I'll look into this tonight or tomorrow.

Environment:

• Dask version:

In [2]: dask.__version__, distributed.__version__
Out[2]: ('2025.3.0+25.g14f5a90e5', '2025.3.0+14.g3f17ddba')

• Python version:
• Operating System:
• Install method (conda, pip, source):

[TomAugspurger]

Initially, this looks a bit like #8998. If we look at the dask graph, we see an Alias without a valid target.

In [12]: dict(result.__dask_graph__())
Out[12]: 
{('from-value-9ee89f5c2a153283d6eaaaebe5281d13',
  0,
  0): Alias(('from-value-9ee89f5c2a153283d6eaaaebe5281d13', 0, 0)->'ndarray-1f263e8315a5ccfd9d7f5487756b70c9')}

Prior to dask/dask@0194ef4, that looked like

In [8]: dict(result.__dask_graph__())
Out[8]: 
{'Future-3176d33b-2935-4c12-804c-ff3b538b412c': <Future: finished, type: numpy.ndarray, key: ndarray-1f263e8315a5ccfd9d7f5487756b70c9>,
 ('from-value-01f8e7801fc4ac85949da73b2552fb1c',
  0,
  0): 'Future-3176d33b-2935-4c12-804c-ff3b538b412c'}