Fix async deadlock risks: use zarr built-in members(), run_in_executor safety

- Replace custom _iter_zarr_groups_async (~90 lines) with zarr's
  AsyncGroup.members(max_depth=None) to avoid sync fallback deadlock
- Wrap _get_open_params and _build_group_members in run_in_executor
  in open_store_async to prevent reentrant sync() deadlock
- Add dedicated executor to _maybe_create_default_indexes_async and
  create_indexes_async to avoid thread pool exhaustion on zarr's IO loop

Author: aladinor

xarray/backends/api.py

@@ -361,21 +361,31 @@ def _datatree_from_backend_datatree(
 
             async def create_indexes_async() -> dict[str, Dataset]:
                 import asyncio
+                from concurrent.futures import ThreadPoolExecutor
 
-                results: dict[str, Dataset] = {}
-                tasks = [
-                    _create_index_for_node(path, node.dataset)
-                    for path, [node] in group_subtrees(backend_tree)
-                ]
-                for fut in asyncio.as_completed(tasks):
-                    path, ds = await fut
-                    results[path] = ds
-                return results
-
-            async def _create_index_for_node(
-                path: str, ds: Dataset
-            ) -> tuple[str, Dataset]:
-                return path, await _maybe_create_default_indexes_async(ds)
+                executor = ThreadPoolExecutor(
+                    max_workers=10, thread_name_prefix="xarray-idx"
+                )
+                try:
+                    results: dict[str, Dataset] = {}
+
+                    async def _create_index_for_node(
+                        path: str, ds: Dataset
+                    ) -> tuple[str, Dataset]:
+                        return path, await _maybe_create_default_indexes_async(
+                            ds, executor=executor
+                        )
+
+                    tasks = [
+                        _create_index_for_node(path, node.dataset)
+                        for path, [node] in group_subtrees(backend_tree)
+                    ]
+                    for fut in asyncio.as_completed(tasks):
+                        path, ds = await fut
+                        results[path] = ds
+                    return results
+                finally:
+                    executor.shutdown(wait=True, cancel_futures=True)
 
             results = zarr_sync(create_indexes_async())
             tree = DataTree.from_dict(results, name=backend_tree.name)
@@ -417,12 +427,10 @@ async def _create_index_for_node(
     return tree
 
 
-async def _maybe_create_default_indexes_async(ds: Dataset) -> Dataset:
-    """Create default indexes for dimension coordinates asynchronously.
+async def _maybe_create_default_indexes_async(
+    ds: Dataset, executor=None
+) -> Dataset:
 
-    This function parallelizes both data loading and index creation,
-    which can significantly speed up opening datasets with many coordinates.
-    """
     import asyncio
 
     to_index_names = [
@@ -434,11 +442,13 @@ async def _maybe_create_default_indexes_async(ds: Dataset) -> Dataset:
     if not to_index_names:
         return ds
 
+    loop = asyncio.get_running_loop()
+
     async def load_var(var: Variable) -> Variable:
         try:
             return await var.load_async()
         except NotImplementedError:
-            return await asyncio.to_thread(var.load)
+            return await loop.run_in_executor(executor, var.load)
 
     await asyncio.gather(*[load_var(ds.variables[name]) for name in to_index_names])