distributed zarr write fail - OSError: too many open files; P2PConsistencyError: No active shuffle with

**Describe the issue**:
I'm processing NetCDF and converting them to Zarr with xarray. For this, I'm using a coiled cluster, dask, xarray, s3fs...

As a normal user who just wants to process data, I'm ending up with random dask behaviour; sometimes, the processing works (rarely), but most of the time it fails with various race conditions. I change vm_types, nthreads, max_pool_connections... heaps of things which don't lead to any sort of success and the log errors I get are all but useful.

**Minimal Complete Verifiable Example**:
```json
"coiled_cluster_options": {
"n_workers": [
20,
100
],
"scheduler_vm_types": "m7i-flex.large",
"worker_vm_types": "m7i-flex.xlarge",
"allow_ingress_from": "me",
"compute_purchase_option": "spot_with_fallback",
"worker_options": {
"nthreads": 2,
}
},
```
I'm either using p2p, or tasks, but end up with the same behaviour
```python

dask.config.set(
{
"array.slicing.split_large_chunks": False,
"distributed.scheduler.worker-saturation": "inf",
"dataframe.shuffle.method": "p2p",
}
)
```
and my distributed file

```yaml
scheduler:
work-stealing: False
allowed-failures: 1 # fail fast
worker:
memory:
spill: False
pause: False
terminate: False
```

Also tried to change the spill, pause, terminate to values such as .90 without any improvements.

The log outputs I get are not human readable,
```
\x00\x00\x00\x00\x00\x00\x8c\x16tblib.pickling_support\x94\x8c\x1dunpickle_exception_with_attrs\x94\x93\x94(\x8c\x08builtins\x94\x8c\x0cRuntimeError\x94\x93\x94}\x94(\x8c\x08__dict__\x94}\x94\x8c\x04args\x94\x8c\xfaError during deserialization of the task graph. This frequently\noccurs if the Scheduler and Client have different environments.\nFor more information, see\nhttps://docs.dask.org/en/stable/deployment-considerations.html#consistent-software-environments\n\x94\x85\x94uh\x00\x8c\x12unpickle_exception\x94\x93\x94(\x8c\x13botocore.exceptions\x94\x8c\x1b_exception_from_packed_args\x94\x93\x94h\x0e\x8c\x17EndpointConnectionError\x94\x93\x94N}\x94(\x8c\x0cendpoint_url\x94\x8c\x90https://imos-data.s3.ap-southeast-2.amazonaws.com/IMOS/SRS/SST/ghrsst/L3SM-1d/dn/2012/20120430092000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn.nc\x94\x8c\x05error\x94h\x02(\x8c\x19aiohttp.client_exceptions\x94\x8c\x17ClientConnectorDNSError\x94\x93\x94}\x94(h\x07}\x94(\x8c\t_conn_key\x94\x8c\x15aiohttp.client_reqrep\x94\x8c\rConnectionKey\x94\x93\x94(\x8c)imos-data.s3.ap-southeast-2.amazonaws.com\x94M\xbb\x01\x88\x88NNNt\x94\x81\x94\x8c\t_os_error\x94h\r(h\x03\x8c\x07OSError\x94\x93\x94K\x18\x8c\x13Too many open files\x94\x86\x94Nh\x00\x8c\x12unpickle_traceback\x94\x93\x94\x8c\x05tblib\x94\x8c\x05Frame\x94\x93\x94)\x81\x94}\x94(\x8c\x08f_locals\x94}\x94\x8c\tf_globals\x94}\x94(\x8c\x08__name__\x94\x8c\x11aiohttp.connector\x94\x8c\x08__file__\x94\x8cA/opt/coiled/env/lib/python3.12/site-packages/aiohttp/connector.py\x94u\x8c\x06f_code\x94h*\x8c\x04Code\x94\x93\x94)\x81\x94}\x94(\x8c\x0bco_filename\x94h6\x8c\x07co_name\x94\x8c\x19_create_direct_connection\x94\x8c\x0bco_argcount\x94K\x00\x8c\x11co_kwonlyargcount\x94K\x00\x8c\x0bco_varnames\x94)\x8c\nco_nlocals\x94K\x00\x8c\x0cco_stacksize\x94K\x00\x8c\x08co_flags\x94K@\x8c\x0eco_firstlineno\x94K\x00ub\x8c\x08f_lineno\x94M\x02\x06ubM\xfc\x05h*\x8c\tTraceback\x94\x93\x94)\x81\x94}\x94(\x8c\x08tb_frame\x94h,)\x81\x94}\x94(h/}\x94h1}\x94(h3h4h5h6uh7h9)\x81\x94}\x94(h/opt/coiled/env/lib/python3.12/site-packages/aiohttp/client.py\x94uh7h9)\x81\x94}\x94(hh?K\x00h@K\x00hA)hBK\x00hCK\x00hDK@hEK\x00ubhFM\x02\x06ubhSM\x02\x06ububububub\x87\x94R\x94h\x8c\x88N)t\x94R\x94h\x1bbu\x87\x94Nh)h,)\x81\x94}\x94(h/}\x94h1}\x94(h3\x8c\x15distributed.scheduler\x94h5\x8cE/opt/coiled/env/lib/python3.12/site-packages/distributed/scheduler.py\x94uh7h9)\x81\x94}\x94(h/opt/coiled/env/lib/python3.12/site-packages/aiohttp/client.py\x94uh7h9)\x81\x94}\x94(hh?K\x00h@K\x00hA)hBK\x00hCK\x00hDK@hEK\x00ubhFM\x02\x06ubhSM\x02\x06ububububub\x87\x94R\x94h\x8c\x88N)t\x94R\x94h\x1bbu\x87\x94Nh)h,)\x81\x94}\x94(h/}\x94h1}\x94(h3\x8c\x15distributed.scheduler\x94h5\x8cE/opt/coiled/env/lib/python3.12/site-packages/distributed/scheduler.py\x94uh7h9)\x81\x94}\x94(hM\xf7\x02\x8c\x07tb_next\x94h3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hG\x8c\x19distributed.shuffle._core\x94hI\x8cI/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_core.py\x94uh"h$)\x81\x94}\x94(h\'hah(\x8c\x0bp2p_barrier\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1MB\x02ubh>M>\x02hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hG\x8c"distributed.shuffle._worker_plugin\x94hI\x8cR/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_worker_plugin.py\x94uh"h$)\x81\x94}\x94(h\'hlh(\x8c\x07barrier\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1M\x87\x01ubh>M\x87\x01hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hG\x8c\x11distributed.utils\x94hI\x8cA/opt/coiled/env/lib/python3.12/site-packages/distributed/utils.py\x94uh"h$)\x81\x94}\x94(h\'hwh(\x8c\x04sync\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1M\xc4\x01ubh>M\xc4\x01hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hGhvhIhwuh"h$)\x81\x94}\x94(h\'hwh(\x8c\x01f\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1M\xae\x01ubh>M\xaa\x01hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hG\x8c\x0btornado.gen\x94hI\x8c;/opt/coiled/env/lib/python3.12/site-packages/tornado/gen.py\x94uh"h$)\x81\x94}\x94(h\'h\x8bh(\x8c\x03run\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1M6\x03ubh>M\x0f\x03hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hGhkhIhluh"h$)\x81\x94}\x94(h\'hlh(\x8c\x08_barrier\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1Mj\x01ubh>Mj\x01hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hGhkhIhluh"h$)\x81\x94}\x94(h\'hlh(\x8c\x0fget_most_recent\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1K\xb1ubh>K\xb1hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hGhkhIhluh"h$)\x81\x94}\x94(h\'hlh(\x8c\x0fget_with_run_id\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1Kwubh>KwhYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hGhkhIhluh"h$)\x81\x94}\x94(h\'hlh(\x8c\x08_refresh\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1K\xdeubh>K\xdehYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hGhkhIhluh"h$)\x81\x94}\x94(h\'hlh(\x8c\x06_fetch\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1K\xc8ubh>K\xc8hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(\x8c\x08__name__\x94\x8c%distributed.shuffle._scheduler_plugin\x94\x8c\x08__file__\x94\x8cU/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_scheduler_plugin.py\x94uh"h$)\x81\x94}\x94(h\'h\xc5h(\x8c\x03get\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1K\xb2ubh>K\xb2ubububububububububububub\x87\x94R\x94hB\x88N)t\x94R\x94h\x08b.'. Traceback (most recent call last): File "/home/ubuntu/github_repo/aodn_cloud_optimised/aodn_cloud_optimised/lib/GenericZarrHandler.py", line 1007, in publish_cloud_optimised_fileset_batch self._write_ds(ds, idx) File "/home/ubuntu/github_repo/aodn_cloud_optimised/aodn_cloud_optimised/lib/GenericZarrHandler.py", line 1786, in _write_ds self._append_zarr_store(ds) File "/home/ubuntu/github_repo/aodn_cloud_optimised/aodn_cloud_optimised/lib/GenericZarrHandler.py", line 1840, in _append_zarr_store ds.to_zarr( File "/home/ubuntu/miniforge3/envs/AodnCloudOptimised/lib/python3.12/site-packages/xarray/core/dataset.py", line 2292, in to_zarr return to_zarr( # type: ignore[call-overload,misc] ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/ubuntu/miniforge3/envs/AodnCloudOptimised/lib/python3.12/site-packages/xarray/backends/api.py", line 2246, in to_zarr writes = writer.sync( ^^^^^^^^^^^^ File "/home/ubuntu/miniforge3/envs/AodnCloudOptimised/lib/python3.12/site-packages/xarray/backends/common.py", line 357, in sync delayed_store = chunkmanager.store( ^^^^^^^^^^^^^^^^^^^ File "/home/ubuntu/miniforge3/envs/AodnCloudOptimised/lib/python3.12/site-packages/xarray/namedarray/daskmanager.py", line 247, in store return store( ^^^^^^ File "/home/ubuntu/miniforge3/envs/AodnCloudOptimised/lib/python3.12/site-packages/dask/array/core.py", line 1221, in store dask.compute(arrays, **kwargs) File "/home/ubuntu/miniforge3/envs/AodnCloudOptimised/lib/python3.12/site-packages/dask/base.py", line 681, in compute results = schedule(expr, keys, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_core.py", line 574, in p2p_barrier File "/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_worker_plugin.py", line 391, in barrier File "/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_worker_plugin.py", line 362, in _barrier File "/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_worker_plugin.py", line 177, in get_most_recent File "/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_worker_plugin.py", line 119, in get_with_run_id File "/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_worker_plugin.py", line 222, in _refresh File "/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_worker_plugin.py", line 200, in _fetch File "/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_scheduler_plugin.py", line 178, in get Exception: b'\x80\x05\x95\x92\x0b\x00\x00\x00\x00\x00\x00\x8c\x16tblib.pickling_support\x94\x8c\x1dunpickle_exception_with_attrs\x94\x93\x94(\x8c\x1fdistributed.shuffle._exceptions\x94\x8c\x13P2PConsistencyError\x94\x93\x94}\x94(\x8c\x08__dict__\x94}\x94\x8c\x04args\x94\x8cBNo active shuffle with id=\'dbe46e5700b3cd9c0e51aa5b1ec8602d\' found\x94\x85\x94uh\x02(\x8c\x08builtins\x94\x8c\x08KeyError\x94\x93\x94}\x94(h\x07}\x94h\t\x8c dbe46e5700b3cd9c0e51aa5b1ec8602d\x94\x85\x94uNh\x00\x8c\x12unpickle_traceback\x94\x93\x94\x8c\x05tblib\x94\x8c\x05Frame\x94\x93\x94)\x81\x94}\x94(\x8c\x08f_locals\x94}\x94\x8c\tf_globals\x94}\x94(\x8c\x08__name__\x94\x8c%distributed.shuffle._scheduler_plugin\x94\x8c\x08__file__\x94\x8cU/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_scheduler_plugin.py\x94u\x8c\x06f_code\x94h\x15\x8c\x04Code\x94\x93\x94)\x81\x94}\x94(\x8c\x0bco_filename\x94h!\x8c\x07co_name\x94\x8c\x03get\x94\x8c\x0bco_argcount\x94K\x00\x8c\x11co_kwonlyargcount\x94K\x00\x8c\x0bco_varnames\x94)\x8c\nco_nlocals\x94K\x00\x8c\x0cco_stacksize\x94K\x00\x8c\x08co_flags\x94K@\x8c\x0eco_firstlineno\x94K\x00ub\x8c\x08f_lineno\x94K\xafubK\xafh\x15\x8c\tTraceback\x94\x93\x94)\x81\x94}\x94(\x8c\x08tb_frame\x94h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(h\x1eh\x1fh h!uh"h$)\x81\x94}\x94(h\'h!h(\x8c\x04_get\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1K\xbeub\x8c\ttb_lineno\x94K\xbeub\x87\x94R\x94N\x89N)t\x94R\x94h\x10bh\x14h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(\x8c\x08__name__\x94\x8c\x12distributed.worker\x94\x8c\x08__file__\x94\x8cB/opt/coiled/env/lib/python3.12/site-packages/distributed/worker.py\x94uh"h$)\x81\x94}\x94(h\'hJh(\x8c\x10_run_task_simple\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1M\xb7\x0bubM\xaa\x0bh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hG\x8c\x0fdask._task_spec\x94hI\x8c?/opt/coiled/env/lib/python3.12/site-packages/dask/_task_spec.py\x94uh"h$)\x81\x9}\x94(h\'hUh(\x8c\x08__call__\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1M\xf7\x02ubh>M\xf7\x02\x8c\x07tb_next\x94h3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hG\x8c\x19distributed.shuffle._core\x94hI\x8cI/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_core.py\x94uh"h$)\x81\x94}\x94(h\'hah(\x8c\x0bp2p_barrier\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1MB\x02ubh>M>\x02hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hG\x8c"distributed.shuffle._worker_plugin\x94hI\x8cR/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_worker_plugin.py\x94uh"h$)\x81\x94}\x94(h\'hlh(\x8c\x07barrier\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1M\x87\x01ubh>M\x87\x01hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hG\x8c\x11distributed.utils\x94hI\x8cA/opt/coiled/env/lib/python3.12/site-packages/distributed/utils.py\x94uh"h$)\x81\x94}\x94(h\'hwh(\x8c\x04sync\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1M\xc4\x01ubh>M\xc4\x01hYh
3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hGhvhIhwuh"h$)\x81\x94}\x94(h\'hwh(\x8c\x01f\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1M\xae\x01ubh>M\xaa\x01hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hG\x8c\x
0btornado.gen\x94hI\x8c;/opt/coiled/env/lib/python3.12/site-packages/tornado/gen.py\x94uh"h$)\x81\x94}\x94(h\'h\x8bh(\x8c\x03run\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1M6\x03ubh>M\x0f\x03hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\
x94h\x1c}\x94(hGhkhIhluh"h$)\x81\x94}\x94(h\'hlh(\x8c\x08_barrier\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1Mj\x01ubh>Mj\x01hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hGhkhIhluh"h$)\x81\x94}\x94(h\'hlh(\x8c\x0fget_most_
recent\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1K\xb1ubh>K\xb1hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hGhkhIhluh"h$)\x81\x94}\x94(h\'hlh(\x8c\x0fget_with_run_id\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1Kwubh
>KwhYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hGhkhIhluh"h$)\x81\x94}\x94(h\'hlh(\x8c\x08_refresh\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1K\xdeubh>K\xdehYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(hGh
khIhluh"h$)\x81\x94}\x94(h\'hlh(\x8c\x06_fetch\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1K\xc8ubh>K\xc8hYh3)\x81\x94}\x94(h6h\x17)\x81\x94}\x94(h\x1a}\x94h\x1c}\x94(\x8c\x08__name__\x94\x8c%distributed.shuffle._scheduler_plugin\x94\x8c\x0
8__file__\x94\x8cU/opt/coiled/env/lib/python3.12/site-packages/distributed/shuffle/_scheduler_plugin.py\x94uh"h$)\x81\x94}\x94(h\'h\xc5h(\x8c\x03get\x94h*K\x00h+K\x00h,)h-K\x00h.K\x00h/K@h0K\x00ubh1K\xb2ubh>K\xb2ubububububububububububub\x87\x94R
\x94hB\x88N)t\x94R\x94h\x08b.'
```

The only way I can digest this is to use an AI.

With coiled dashboard, I can see that most of the time, none of the workers/scheduler have any sort of error, good mem/cpu usage.

**Anything else we need to know?**:

**Environment**:

- Dask version: distributed 2025.10.0
- Python version:
- Operating System:
- Install method (conda, pip, source):