better lost deps log msg

Author: fjetter

distributed/client.py

@@ -3763,12 +3763,12 @@ def persist(
         metadata = SpanMetadata(
             collections=[get_collections_metadata(v) for v in collections]
         )
-        dsk = collections_to_expr(collections, optimize_graph, **kwargs)
+        expr = collections_to_expr(collections, optimize_graph, **kwargs)
 
         names = {k for c in collections for k in flatten(c.__dask_keys__())}
 
         futures = self._graph_to_futures(
-            dsk,
+            expr,
             names,
             workers=workers,
             allow_other_workers=allow_other_workers,
@@ -6087,7 +6087,8 @@ def futures_of(o, client=None):
         elif isinstance(x, TaskRef):
             if x not in seen:
                 seen.add(x)
-                futures.append(x)
+                if isinstance(x, Future):
+                    futures.append(x)
         elif dask.is_dask_collection(x):
             stack.extend(x.__dask_graph__().values())
 

distributed/scheduler.py

@@ -4617,7 +4617,11 @@ def _find_lost_dependencies(
                     if d not in self.tasks:
                         lost_keys.add(d)
                         lost_keys.add(k)
-                        logger.info("User asked for computation on lost data, %s", k)
+                        logger.info(
+                            "User asked for computation on lost data. Final key %s is missing dependency %s",
+                            k,
+                            d,
+                        )
                         dependencies.pop(d, None)
                         keys.discard(k)
                     continue