@@ -3,7 +3,7 @@
 import copy
 import math
 import warnings
-from asyncio import gather
+from asyncio import TaskGroup, gather
 from collections.abc import Iterable, Mapping, Sequence
 from dataclasses import dataclass, field, replace
 from itertools import starmap
@@ -5940,7 +5940,15 @@ async def _delete_worker() -> None:
         # Bound the worker count even when the user has disabled the I/O
         # concurrency limit with None.
         concurrency = zarr_config.get("async.concurrency") or 1000
-        await gather(*(_delete_worker() for _ in range(concurrency)))
+        # A TaskGroup cancels the remaining workers on the first failure, so no
+        # deletions continue after resize has raised.
+        try:
+            async with TaskGroup() as tg:
+                for _ in range(concurrency):
+                    tg.create_task(_delete_worker())
+        except ExceptionGroup as eg:
+            # Keep resize's contract of raising the store's own exception.
+            raise eg.exceptions[0] from None
 
     # Write new metadata
     await save_metadata(array.store_path, new_metadata)
 
@@ -155,3 +155,29 @@ async def test_resize_delete_failure_preserves_metadata() -> None:
     assert metadata_before.to_bytes() == metadata_after.to_bytes()
     assert array.shape == (4,)
     assert array._chunk_grid is grid_before
+
+
+async def test_resize_delete_failure_stops_remaining_deletions() -> None:
+    store = MemoryStore()
+    array = await create_array(store, shape=(3000,), chunks=(1,), dtype="uint8")
+    calls = 0
+
+    async def delete(key: str) -> None:
+        nonlocal calls
+        calls += 1
+        call = calls
+        await asyncio.sleep(0)
+        if call == 10:
+            raise OSError("delete failed")
+
+    with (
+        config.set({"async.concurrency": 4}),
+        mock.patch.object(store, "delete", delete),
+    ):
+        with pytest.raises(OSError, match="delete failed"):
+            await array.resize((0,))
+        calls_at_raise = calls
+        # Give any surviving workers a chance to run.
+        for _ in range(10):
+            await asyncio.sleep(0)
+    assert calls == calls_at_raise < 3000