Add failing test for forgotten tasks

distributed/shuffle/tests/test_shuffle.py

@@ -395,3 +395,30 @@ async def test_restrictions(c, s, a, b):
 
     await y
     assert all(stringify(key) in a.data for key in y.__dask_keys__())
+
+
+@pytest.mark.xfail(reason="Don't clean up forgotten shuffles")
+@gen_cluster(client=True)
+async def test_delete_some_results(c, s, a, b):
+    df = dask.datasets.timeseries(
+        start="2000-01-01",
+        end="2000-01-10",
+        dtypes={"x": float, "y": float},
+        freq="10 s",
+    )
+    x = dd.shuffle.shuffle(df, "x", shuffle="p2p").persist()
+    while not s.tasks or not any(ts.state == "memory" for ts in s.tasks.values()):
+        await asyncio.sleep(0.01)
+
+    n = len(s.tasks)
+
+    x = x.partitions[: x.npartitions // 2].persist()
+
+    while len(s.tasks) == n:
+        await asyncio.sleep(0.1)
+
+    await x
+
+    clean_worker(a)
+    clean_worker(b)
+    clean_scheduler(s)