From 8b62f91e1f546075ff5a4f9678c7f117a9f0a941 Mon Sep 17 00:00:00 2001
From: Davis Vann Bennett <davis.v.bennett@gmail.com>
Date: Fri, 25 Sep 2026 09:30:35 +0200
Subject: [PATCH 1/3] fix: validate a new node before overwriting the existing
 one

Opening or creating a node with mode="w" or overwrite=True deleted the
existing node before the new one was validated, so an invalid argument
destroyed the existing data and created nothing. For example,
zarr.open(store, mode="w", config=...) on an existing array erased the
store and then raised TypeError from open_group.

Two deletion points ran too early:
- make_store_path(mode="w") erased the path before dispatch in open,
  open_array, open_group, create and save_array. These now build the
  store path with mode "a" and create the node with overwrite=True.
- overwrite=True deleted the node before its metadata was built in
  AsyncArray._create_v2/_create_v3, init_array and AsyncGroup.from_store.
  save_new_metadata now encodes the metadata first, then deletes, then
  writes.

save_group checks its arguments before erasing the path.

Assisted-by: ClaudeCode:claude-opus-5-5
---
 src/zarr/api/asynchronous.py | 102 ++++++++++++++++++++-----------
 src/zarr/core/array.py       |  32 ++--------
 src/zarr/core/group.py       |  16 ++---
 src/zarr/core/metadata/io.py |  45 +++++++++++++-
 tests/test_api.py            | 112 +++++++++++++++++++++++++++++++++++
 5 files changed, 234 insertions(+), 73 deletions(-)

diff --git a/src/zarr/api/asynchronous.py b/src/zarr/api/asynchronous.py
index 82300b17f2..9aba516866 100644
--- a/src/zarr/api/asynchronous.py
+++ b/src/zarr/api/asynchronous.py
@@ -108,6 +108,18 @@ def _infer_overwrite(mode: AccessModeLiteral) -> bool:
     return mode in _OVERWRITE_MODES
 
 
+def _deferred_overwrite_mode(mode: AccessModeLiteral) -> AccessModeLiteral:
+    """
+    The mode for building the store path of a node that `mode` may overwrite.
+
+    Building a store path with mode `"w"` deletes everything under the path before the new
+    node has been validated, so an invalid argument would destroy the existing node and create
+    nothing. Callers build the store path with mode `"a"` instead and create the node with
+    `overwrite=True`, which deletes the existing node only once the new metadata is valid.
+    """
+    return "a" if mode == "w" else mode
+
+
 def _warn_unimplemented_kwargs(kwargs: dict[str, Any]) -> None:
     """
     Emit a "not yet implemented" warning for each provided keyword argument that is not None.
@@ -401,23 +413,32 @@ async def open(
             mode = "r"
         else:
             mode = "a"
-    store_path = await make_store_path(store, mode=mode, path=path, storage_options=storage_options)
+    store_path = await make_store_path(
+        store,
+        mode=_deferred_overwrite_mode(mode),
+        path=path,
+        storage_options=storage_options,
+    )
 
     # TODO: the mode check below seems wrong!
     if "shape" not in kwargs and mode in {"a", "r", "r+", "w"}:
-        try:
-            metadata_dict = await get_array_metadata(store_path, zarr_format=zarr_format)
-            # TODO: remove this cast when we fix typing for array metadata dicts
-            _metadata_dict = cast("ArrayMetadataDict", metadata_dict)
-            # for v2, the above would already have raised an exception if not an array
-            zarr_format = _metadata_dict["zarr_format"]
-            is_v3_array = zarr_format == 3 and _metadata_dict.get("node_type") == "array"
-            if is_v3_array or zarr_format == 2:
-                return AsyncArray(
-                    store_path=store_path, metadata=_metadata_dict, config=kwargs.get("config")
-                )
-        except (FileNotFoundError, NodeTypeValidationError):
-            pass
+        # mode "w" replaces any existing node, so there is nothing to open
+        if mode != "w":
+            try:
+                metadata_dict = await get_array_metadata(store_path, zarr_format=zarr_format)
+                # TODO: remove this cast when we fix typing for array metadata dicts
+                _metadata_dict = cast("ArrayMetadataDict", metadata_dict)
+                # for v2, the above would already have raised an exception if not an array
+                zarr_format = _metadata_dict["zarr_format"]
+                is_v3_array = zarr_format == 3 and _metadata_dict.get("node_type") == "array"
+                if is_v3_array or zarr_format == 2:
+                    return AsyncArray(
+                        store_path=store_path,
+                        metadata=_metadata_dict,
+                        config=kwargs.get("config"),
+                    )
+            except (FileNotFoundError, NodeTypeValidationError):
+                pass
         return await open_group(store=store_path, zarr_format=zarr_format, mode=mode, **kwargs)
 
     try:
@@ -526,7 +547,9 @@ async def save_array(
         raise TypeError("arr argument must be numpy or other NDArrayLike array")
 
     mode = kwargs.pop("mode", "a")
-    store_path = await make_store_path(store, path=path, mode=mode, storage_options=storage_options)
+    store_path = await make_store_path(
+        store, path=path, mode=_deferred_overwrite_mode(mode), storage_options=storage_options
+    )
     if np.isscalar(arr):
         arr = np.array(arr)
     shape = arr.shape
@@ -575,8 +598,6 @@ async def save_group(
         NumPy arrays with data to save.
     """
 
-    store_path = await make_store_path(store, path=path, mode="w", storage_options=storage_options)
-
     if zarr_format is None:
         zarr_format = _default_zarr_format()
 
@@ -591,6 +612,9 @@ async def save_group(
 
     if len(args) == 0 and len(kwargs) == 0:
         raise ValueError("at least one array must be provided")
+
+    # mode "w" deletes everything under the path, so the arguments are checked first
+    store_path = await make_store_path(store, path=path, mode="w", storage_options=storage_options)
     aws = []
     # `store_path` already consumed `storage_options`, so passing them on again would
     # make `make_store_path` reject them as unused.
@@ -876,7 +900,9 @@ async def open_group(
         }
     )
 
-    store_path = await make_store_path(store, mode=mode, storage_options=storage_options, path=path)
+    store_path = await make_store_path(
+        store, mode=_deferred_overwrite_mode(mode), storage_options=storage_options, path=path
+    )
     if attributes is None:
         attributes = {}
 
@@ -1102,7 +1128,10 @@ async def create(
 
     if mode is None:
         mode = "a"
-    store_path = await make_store_path(store, path=path, mode=mode, storage_options=storage_options)
+    overwrite = overwrite or _infer_overwrite(mode)
+    store_path = await make_store_path(
+        store, path=path, mode=_deferred_overwrite_mode(mode), storage_options=storage_options
+    )
 
     config_parsed = parse_array_config(config)
 
@@ -1316,25 +1345,30 @@ async def open_array(
     """
 
     mode = kwargs.pop("mode", None)
-    store_path = await make_store_path(store, path=path, mode=mode, storage_options=storage_options)
+    store_path = await make_store_path(
+        store,
+        path=path,
+        mode=None if mode is None else _deferred_overwrite_mode(mode),
+        storage_options=storage_options,
+    )
 
     if "write_empty_chunks" in kwargs:
         _warn_write_empty_chunks_kwarg()
 
-    try:
-        return await AsyncArray.open(store_path, zarr_format=zarr_format)
-    except FileNotFoundError as err:
-        if not store_path.read_only and mode in _CREATE_MODES:
-            overwrite = _infer_overwrite(mode)
-            _zarr_format = zarr_format or _default_zarr_format()
-            return await create(
-                store=store_path,
-                zarr_format=_zarr_format,
-                overwrite=overwrite,
-                **kwargs,
-            )
-        msg = f"No array found in store {store_path.store} at path {store_path.path}"
-        raise ArrayNotFoundError(msg) from err
+    # mode "w" replaces any existing array, so there is nothing to open
+    if mode != "w":
+        try:
+            return await AsyncArray.open(store_path, zarr_format=zarr_format)
+        except FileNotFoundError as err:
+            if store_path.read_only or mode not in _CREATE_MODES:
+                msg = f"No array found in store {store_path.store} at path {store_path.path}"
+                raise ArrayNotFoundError(msg) from err
+    return await create(
+        store=store_path,
+        zarr_format=zarr_format or _default_zarr_format(),
+        overwrite=_infer_overwrite(mode),
+        **kwargs,
+    )
 
 
 async def open_like(
diff --git a/src/zarr/core/array.py b/src/zarr/core/array.py
index 133f0ff72c..6e7b8d6ee8 100644
--- a/src/zarr/core/array.py
+++ b/src/zarr/core/array.py
@@ -118,7 +118,7 @@
     ArrayV2MetadataDict,
     ArrayV3Metadata,
 )
-from zarr.core.metadata.io import save_metadata
+from zarr.core.metadata.io import save_metadata, save_new_metadata
 from zarr.core.metadata.v2 import (
     CompressorLikev2,
     get_object_codec_id,
@@ -146,7 +146,7 @@
     _parse_bytes_bytes_codec,
     get_pipeline_class,
 )
-from zarr.storage._common import StorePath, ensure_no_existing_node, make_store_path
+from zarr.storage._common import StorePath, make_store_path
 from zarr.storage._utils import _relativize_path
 
 if TYPE_CHECKING:
@@ -334,22 +334,6 @@ def _array_metadata_dict_v3(zarr_json_bytes: Buffer) -> dict[str, JSON]:
     return metadata_dict
 
 
-async def _prepare_overwrite(
-    store_path: StorePath, *, zarr_format: ZarrFormat, overwrite: bool
-) -> None:
-    """
-    Prepare a store path for writing a new node.
-
-    If `overwrite` is true and the store supports deletes, any existing node at
-    `store_path` is deleted. Otherwise, the absence of an existing node is enforced
-    (raising if one is present).
-    """
-    if overwrite and store_path.store.supports_deletes:
-        await store_path.delete_dir()
-    else:
-        await ensure_no_existing_node(store_path, zarr_format=zarr_format)
-
-
 @dataclass(frozen=True)
 class AsyncArray[T_ArrayMetadata: (ArrayV2Metadata, ArrayV3Metadata)]:
     """
@@ -629,8 +613,6 @@ async def _create_v3(
         attributes: dict[str, JSON] | None = None,
         overwrite: bool = False,
     ) -> AsyncArrayV3:
-        await _prepare_overwrite(store_path, zarr_format=3, overwrite=overwrite)
-
         if isinstance(chunk_key_encoding, tuple):
             chunk_key_encoding = (
                 V2ChunkKeyEncoding(separator=chunk_key_encoding[1])
@@ -650,7 +632,7 @@ async def _create_v3(
         )
 
         array = cls(metadata=metadata, store_path=store_path, config=config)
-        await array._save_metadata(metadata, ensure_parents=True)
+        await save_new_metadata(store_path, metadata, overwrite=overwrite, ensure_parents=True)
         return array
 
     @staticmethod
@@ -704,8 +686,6 @@ async def _create_v2(
         attributes: dict[str, JSON] | None = None,
         overwrite: bool = False,
     ) -> AsyncArrayV2:
-        await _prepare_overwrite(store_path, zarr_format=2, overwrite=overwrite)
-
         compressor_parsed: CompressorLikev2
         if compressor == "auto":
             compressor_parsed = default_compressor_v2(dtype)
@@ -733,7 +713,7 @@ async def _create_v2(
         )
 
         array = cls(metadata=metadata, store_path=store_path, config=config)
-        await array._save_metadata(metadata, ensure_parents=True)
+        await save_new_metadata(store_path, metadata, overwrite=overwrite, ensure_parents=True)
         return array
 
     @classmethod
@@ -4530,8 +4510,6 @@ async def init_array(
         chunk_key_encoding, zarr_format=zarr_format
     )
 
-    await _prepare_overwrite(store_path, zarr_format=zarr_format, overwrite=overwrite)
-
     # Validate rectilinear chunks constraints
     if _is_rectilinear_chunks(chunks):
         if zarr_format == 2:
@@ -4645,7 +4623,7 @@ async def init_array(
         )
 
     arr = AsyncArray(metadata=meta, store_path=store_path, config=config)
-    await arr._save_metadata(meta, ensure_parents=True)
+    await save_new_metadata(store_path, meta, overwrite=overwrite, ensure_parents=True)
     return arr
 
 
diff --git a/src/zarr/core/group.py b/src/zarr/core/group.py
index 1eb18ca3e8..ed4bca6f0d 100644
--- a/src/zarr/core/group.py
+++ b/src/zarr/core/group.py
@@ -49,7 +49,7 @@
 from zarr.core.dtype import parse_data_type
 from zarr.core.json_parse import parse_field
 from zarr.core.metadata import ArrayV2Metadata, ArrayV3Metadata
-from zarr.core.metadata.io import save_metadata
+from zarr.core.metadata.io import save_metadata, save_new_metadata
 from zarr.core.metadata.v3 import AllowedExtraField, parse_extra_fields
 from zarr.core.sync import SyncMixin, sync
 from zarr.errors import (
@@ -62,7 +62,7 @@
     ZarrUserWarning,
 )
 from zarr.storage import StoreLike, StorePath
-from zarr.storage._common import ensure_no_existing_node, make_store_path
+from zarr.storage._common import make_store_path
 from zarr.storage._utils import _join_paths, _normalize_path_keys, normalize_path
 
 if TYPE_CHECKING:
@@ -506,20 +506,14 @@ async def from_store(
         zarr_format: ZarrFormat = 3,
     ) -> AsyncGroup:
         store_path = await make_store_path(store)
-
-        if overwrite:
-            if store_path.store.supports_deletes:
-                await store_path.delete_dir()
-            else:
-                await ensure_no_existing_node(store_path, zarr_format=zarr_format)
-        else:
-            await ensure_no_existing_node(store_path, zarr_format=zarr_format)
         attributes = attributes or {}
         group = cls(
             metadata=GroupMetadata(attributes=attributes, zarr_format=zarr_format),
             store_path=store_path,
         )
-        await group._save_metadata(ensure_parents=True)
+        await save_new_metadata(
+            store_path, group.metadata, overwrite=overwrite, ensure_parents=True
+        )
         return group
 
     @classmethod
diff --git a/src/zarr/core/metadata/io.py b/src/zarr/core/metadata/io.py
index 7b63f5493b..834c385a60 100644
--- a/src/zarr/core/metadata/io.py
+++ b/src/zarr/core/metadata/io.py
@@ -9,6 +9,7 @@
 from zarr.storage._common import StorePath, ensure_no_existing_node
 
 if TYPE_CHECKING:
+    from zarr.core.buffer import Buffer
     from zarr.core.common import ZarrFormat
     from zarr.core.group import GroupMetadata
     from zarr.core.metadata import ArrayMetadata
@@ -52,11 +53,53 @@ async def save_metadata(
     ValueError
     """
     to_save = metadata.to_buffer_dict(default_buffer_prototype())
+    await _write_metadata(store_path, to_save, metadata.zarr_format, ensure_parents=ensure_parents)
+
+
+async def save_new_metadata(
+    store_path: StorePath,
+    metadata: ArrayMetadata | GroupMetadata,
+    *,
+    overwrite: bool,
+    ensure_parents: bool = False,
+) -> None:
+    """Save the metadata of a new array or group, replacing any existing node if requested.
+
+    The metadata is encoded before the store is modified, so metadata that cannot be
+    encoded raises without deleting an existing node.
+
+    Parameters
+    ----------
+    store_path : StorePath
+        Location of the new node.
+    metadata : ArrayMetadata | GroupMetadata
+        Metadata of the new node.
+    overwrite : bool
+        If true and the store supports deletes, delete any existing node at `store_path`.
+        Otherwise, raise if a node already exists at `store_path`.
+    ensure_parents : bool, optional
+        Create any missing parent groups, and check no existing parents are arrays.
+    """
+    to_save = metadata.to_buffer_dict(default_buffer_prototype())
+    if overwrite and store_path.store.supports_deletes:
+        await store_path.delete_dir()
+    else:
+        await ensure_no_existing_node(store_path, zarr_format=metadata.zarr_format)
+    await _write_metadata(store_path, to_save, metadata.zarr_format, ensure_parents=ensure_parents)
+
+
+async def _write_metadata(
+    store_path: StorePath,
+    to_save: dict[str, Buffer],
+    zarr_format: ZarrFormat,
+    *,
+    ensure_parents: bool,
+) -> None:
     set_awaitables = [set_or_delete(store_path / key, value) for key, value in to_save.items()]
 
     if ensure_parents:
         # To enable zarr.create(store, path="a/b/c"), we need to create all the intermediate groups.
-        parents = _build_parents(store_path, metadata.zarr_format)
+        parents = _build_parents(store_path, zarr_format)
         ensure_array_awaitables = []
 
         for parent_path, parent_metadata in parents.items():
diff --git a/tests/test_api.py b/tests/test_api.py
index cc72cd7000..54db82de1f 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -526,6 +526,118 @@ def test_open_like_zarr_format(source_format: ZarrFormat, target_format: ZarrFor
     }
 
 
+def _store_with_array(zarr_format: ZarrFormat) -> MemoryStore:
+    """A MemoryStore holding a written int32 array of shape (4,) at the root."""
+    store = MemoryStore()
+    arr = zarr.create_array(store, shape=(4,), dtype="int32", zarr_format=zarr_format)
+    arr[:] = 1
+    return store
+
+
+def _store_contents(store: MemoryStore) -> dict[str, bytes]:
+    return {key: value.to_bytes() for key, value in store._store_dict.items()}
+
+
+# Each call replaces the node at the root of the store it is given.
+_OVERWRITE_CALLS: dict[str, Callable[..., AnyArray | Group]] = {
+    "open-array": lambda store, **kw: zarr.open(store=store, mode="w", shape=(2,), **kw),
+    "open-group": lambda store, **kw: zarr.open(store=store, mode="w", **kw),
+    "open_array": lambda store, **kw: zarr.open_array(store=store, mode="w", shape=(2,), **kw),
+    "open_group": lambda store, **kw: zarr.open_group(store=store, mode="w", **kw),
+    "create-mode": lambda store, **kw: zarr.create(store=store, mode="w", shape=(2,), **kw),
+    "create-overwrite": lambda store, **kw: zarr.create(
+        store=store, overwrite=True, shape=(2,), **kw
+    ),
+    "create_array": lambda store, **kw: zarr.create_array(
+        store, overwrite=True, shape=(2,), dtype="int32", **kw
+    ),
+    "save_array": lambda store, **kw: zarr.save_array(store, np.arange(2), mode="w", **kw),
+    "group": lambda store, **kw: zarr.group(store=store, overwrite=True, **kw),
+}
+
+
+@pytest.mark.parametrize("zarr_format", [2, 3])
+@pytest.mark.parametrize("call", _OVERWRITE_CALLS)
+def test_overwrite_replaces_existing_node(zarr_format: ZarrFormat, call: str) -> None:
+    """
+    Opening or creating a node in an overwriting mode replaces the node at the path,
+    including its chunks.
+    """
+    store = _store_with_array(zarr_format)
+    _OVERWRITE_CALLS[call](store)
+    reopened = zarr.open(store=store, mode="r")
+    if call in {"open-group", "open_group", "group"}:
+        assert isinstance(reopened, Group)
+        assert set(store._store_dict) <= {"zarr.json", ".zgroup", ".zattrs"}
+    else:
+        assert isinstance(reopened, Array)
+        assert reopened.shape == (2,)
+
+
+@pytest.mark.parametrize(
+    "call",
+    ["open-array", "open-group", "open_array", "create-mode", "create-overwrite", "save_array"],
+)
+def test_overwrite_invalid_keyword_keeps_existing_node(call: str) -> None:
+    """
+    An overwriting call that is given a keyword argument its target cannot accept raises
+    without modifying the existing node.
+    """
+    store = _store_with_array(3)
+    before = _store_contents(store)
+    kwargs = {"config": {"order": "F"}} if call == "open-group" else {"bogus": 1}
+    with pytest.raises(TypeError, match="unexpected keyword argument"):
+        _OVERWRITE_CALLS[call](store, **kwargs)
+    assert _store_contents(store) == before
+
+
+@pytest.mark.parametrize(
+    "call",
+    ["open-array", "open_array", "create-mode", "create-overwrite", "create_array", "save_array"],
+)
+def test_overwrite_invalid_fill_value_keeps_existing_node(call: str) -> None:
+    """
+    An overwriting call whose new array metadata cannot be built raises without modifying
+    the existing node.
+    """
+    store = _store_with_array(3)
+    before = _store_contents(store)
+    kwargs: dict[str, Any] = {"fill_value": "not a number"}
+    if call in {"open-array", "open_array", "create-mode", "create-overwrite"}:
+        kwargs["dtype"] = "int32"
+    with pytest.raises(ValueError, match="invalid literal"):
+        _OVERWRITE_CALLS[call](store, **kwargs)
+    assert _store_contents(store) == before
+
+
+@pytest.mark.parametrize(
+    "call",
+    ["open-array", "open-group", "open_array", "open_group", "create_array", "group"],
+)
+def test_overwrite_unencodable_attributes_keeps_existing_node(call: str) -> None:
+    """
+    An overwriting call whose new metadata cannot be encoded raises without modifying the
+    existing node.
+    """
+    store = _store_with_array(3)
+    before = _store_contents(store)
+    with pytest.raises(TypeError, match="not JSON serializable"):
+        _OVERWRITE_CALLS[call](store, attributes={"x": object()})
+    assert _store_contents(store) == before
+
+
+def test_save_group_invalid_argument_keeps_existing_node() -> None:
+    """
+    save_group given an argument that is not an array raises without modifying the
+    existing node.
+    """
+    store = _store_with_array(3)
+    before = _store_contents(store)
+    with pytest.raises(TypeError, match="must be a numpy or other NDArrayLike array"):
+        zarr.save_group(store, a=np.arange(2), b="not an array")  # type: ignore[arg-type]
+    assert _store_contents(store) == before
+
+
 @pytest.mark.parametrize("store", ["memory", "local", "zip"], indirect=True)
 def test_v2_and_v3_exist_at_same_path(store: Store) -> None:
     zarr.create_array(store, shape=(10,), dtype="uint8", zarr_format=3)

From b5705f1f591a6caee8045da9e8409d67a0b9fd39 Mon Sep 17 00:00:00 2001
From: Davis Vann Bennett <davis.v.bennett@gmail.com>
Date: Fri, 25 Sep 2026 09:31:11 +0200
Subject: [PATCH 2/3] docs(changes): add fragment for #4410

Assisted-by: ClaudeCode:claude-opus-5-5
---
 changes/4410.bugfix.md | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 changes/4410.bugfix.md

diff --git a/changes/4410.bugfix.md b/changes/4410.bugfix.md
new file mode 100644
index 0000000000..0a2e3ee009
--- /dev/null
+++ b/changes/4410.bugfix.md
@@ -0,0 +1 @@
+Opening or creating an array or group with `mode="w"` or `overwrite=True` no longer deletes the existing node when the call fails. Previously the existing node was deleted before the new one was validated, so an invalid argument (for example `zarr.open(store, mode="w", config=...)`, an invalid `fill_value`, or attributes that are not JSON) destroyed the existing data and created nothing.

From 2531f58555da04a9529e2bbe96234f62d99b2273 Mon Sep 17 00:00:00 2001
From: Davis Vann Bennett <davis.v.bennett@gmail.com>
Date: Wed, 30 Sep 2026 19:53:32 +0200
Subject: [PATCH 3/3] fix(storage): let node creation own the mode "w" erase

`StorePath.open(mode="w")` no longer deletes the keys under the path. The
node created there replaces them through `save_new_metadata(overwrite=True)`,
once its metadata is valid, so the API functions pass their mode through
unchanged and `_deferred_overwrite_mode` is gone. A read-only store refused
with mode "w" names that mode again in its error.

`save_group` resolves every array's data type before the group is replaced,
and replaces it explicitly with `AsyncGroup.from_store(overwrite=True)`.

The docstrings of `open`, `open_group` and `create` state that a store which
cannot delete keys raises instead of replacing an existing node; the tests
pin that contract over `ZipStore`, seed the overwrite tests with a group that
has children, and cover the read-only error and the `save_group` dtype path.

Assisted-by: ClaudeCode:claude-fable-5-1
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 changes/4410.bugfix.md        |   2 +-
 src/zarr/api/asynchronous.py  |  60 ++++++---------
 src/zarr/api/synchronous.py   |   9 ++-
 src/zarr/core/metadata/io.py  |   5 +-
 src/zarr/storage/_common.py   |  22 +++---
 tests/test_api.py             | 133 ++++++++++++++++++++++++++++------
 tests/test_store/test_core.py |  11 +++
 7 files changed, 161 insertions(+), 81 deletions(-)

diff --git a/changes/4410.bugfix.md b/changes/4410.bugfix.md
index 0a2e3ee009..2fe2981d03 100644
--- a/changes/4410.bugfix.md
+++ b/changes/4410.bugfix.md
@@ -1 +1 @@
-Opening or creating an array or group with `mode="w"` or `overwrite=True` no longer deletes the existing node when the call fails. Previously the existing node was deleted before the new one was validated, so an invalid argument (for example `zarr.open(store, mode="w", config=...)`, an invalid `fill_value`, or attributes that are not JSON) destroyed the existing data and created nothing.
+Opening or creating an array or group with `mode="w"` or `overwrite=True` no longer deletes the existing node when the call fails. Previously the existing node was deleted before the new one was validated, so an invalid argument (for example `zarr.open(store, mode="w", config=...)`, an invalid `fill_value`, attributes that are not JSON, or an array `zarr.save_group` cannot store) destroyed the existing data and created nothing. Deletion now belongs to node creation: `StorePath.open(mode="w")` no longer erases the keys under the path, and the node written there replaces them once its metadata is valid. On a store that cannot delete keys, such as `ZipStore`, `mode="w"` creates the node when the path is empty and raises `ContainsArrayError` or `ContainsGroupError` when a node already exists, instead of `NotImplementedError`.
diff --git a/src/zarr/api/asynchronous.py b/src/zarr/api/asynchronous.py
index 9aba516866..e8d5e5d631 100644
--- a/src/zarr/api/asynchronous.py
+++ b/src/zarr/api/asynchronous.py
@@ -108,18 +108,6 @@ def _infer_overwrite(mode: AccessModeLiteral) -> bool:
     return mode in _OVERWRITE_MODES
 
 
-def _deferred_overwrite_mode(mode: AccessModeLiteral) -> AccessModeLiteral:
-    """
-    The mode for building the store path of a node that `mode` may overwrite.
-
-    Building a store path with mode `"w"` deletes everything under the path before the new
-    node has been validated, so an invalid argument would destroy the existing node and create
-    nothing. Callers build the store path with mode `"a"` instead and create the node with
-    `overwrite=True`, which deletes the existing node only once the new metadata is valid.
-    """
-    return "a" if mode == "w" else mode
-
-
 def _warn_unimplemented_kwargs(kwargs: dict[str, Any]) -> None:
     """
     Emit a "not yet implemented" warning for each provided keyword argument that is not None.
@@ -378,7 +366,8 @@ async def open(
         Persistence mode: 'r' means read only (must exist); 'r+' means
         read/write (must exist); 'a' means read/write (create if doesn't
         exist); 'w' means create (overwrite if exists); 'w-' means create
-        (fail if exists).
+        (fail if exists). On a store that cannot delete keys, 'w' raises an
+        error instead of replacing an existing node.
         If the store is read-only, the default is 'r'; otherwise, it is 'a'.
     zarr_format : {2, 3, None}, optional
         The zarr format to use when saving.
@@ -413,17 +402,12 @@ async def open(
             mode = "r"
         else:
             mode = "a"
-    store_path = await make_store_path(
-        store,
-        mode=_deferred_overwrite_mode(mode),
-        path=path,
-        storage_options=storage_options,
-    )
+    store_path = await make_store_path(store, mode=mode, path=path, storage_options=storage_options)
 
     # TODO: the mode check below seems wrong!
-    if "shape" not in kwargs and mode in {"a", "r", "r+", "w"}:
+    if "shape" not in kwargs and mode in (*_READ_MODES, "w"):
         # mode "w" replaces any existing node, so there is nothing to open
-        if mode != "w":
+        if mode in _READ_MODES:
             try:
                 metadata_dict = await get_array_metadata(store_path, zarr_format=zarr_format)
                 # TODO: remove this cast when we fix typing for array metadata dicts
@@ -547,9 +531,7 @@ async def save_array(
         raise TypeError("arr argument must be numpy or other NDArrayLike array")
 
     mode = kwargs.pop("mode", "a")
-    store_path = await make_store_path(
-        store, path=path, mode=_deferred_overwrite_mode(mode), storage_options=storage_options
-    )
+    store_path = await make_store_path(store, path=path, mode=mode, storage_options=storage_options)
     if np.isscalar(arr):
         arr = np.array(arr)
     shape = arr.shape
@@ -613,8 +595,15 @@ async def save_group(
     if len(args) == 0 and len(kwargs) == 0:
         raise ValueError("at least one array must be provided")
 
-    # mode "w" deletes everything under the path, so the arguments are checked first
+    # Resolve every data type before anything is deleted, so an array that Zarr cannot
+    # store raises while the existing node is still intact.
+    for arr in (*args, *kwargs.values()):
+        get_data_type_from_native_dtype(arr.dtype)
+
     store_path = await make_store_path(store, path=path, mode="w", storage_options=storage_options)
+    # The group replaces whatever is stored under the path, now that the arguments are known
+    # to be valid.
+    await AsyncGroup.from_store(store_path, zarr_format=zarr_format, overwrite=True)
     aws = []
     # `store_path` already consumed `storage_options`, so passing them on again would
     # make `make_store_path` reject them as unused.
@@ -847,7 +836,8 @@ async def open_group(
         Persistence mode: 'r' means read only (must exist); 'r+' means
         read/write (must exist); 'a' means read/write (create if doesn't
         exist); 'w' means create (overwrite if exists); 'w-' means create
-        (fail if exists).
+        (fail if exists). On a store that cannot delete keys, 'w' raises an
+        error instead of replacing an existing node.
     cache_attrs : bool, optional
         If True (default), user attributes will be cached for attribute read
         operations. If False, user attributes are reloaded from the store prior
@@ -900,9 +890,7 @@ async def open_group(
         }
     )
 
-    store_path = await make_store_path(
-        store, mode=_deferred_overwrite_mode(mode), storage_options=storage_options, path=path
-    )
+    store_path = await make_store_path(store, mode=mode, storage_options=storage_options, path=path)
     if attributes is None:
         attributes = {}
 
@@ -1096,7 +1084,8 @@ async def create(
           `True`.
         - `'r'` always fails.
 
-        `mode` has no effect, and is not validated, if `store` is a `StorePath`.
+        If `store` is a `StorePath`, `mode` is not validated against it: `'w'` still
+        sets `overwrite`, and the other modes have no effect.
     data : array-like, optional
         Values written into the new array after it is created. Unlike the `data`
         parameter of `create_array`, it does not set the shape or data type of the
@@ -1129,9 +1118,7 @@ async def create(
     if mode is None:
         mode = "a"
     overwrite = overwrite or _infer_overwrite(mode)
-    store_path = await make_store_path(
-        store, path=path, mode=_deferred_overwrite_mode(mode), storage_options=storage_options
-    )
+    store_path = await make_store_path(store, path=path, mode=mode, storage_options=storage_options)
 
     config_parsed = parse_array_config(config)
 
@@ -1345,12 +1332,7 @@ async def open_array(
     """
 
     mode = kwargs.pop("mode", None)
-    store_path = await make_store_path(
-        store,
-        path=path,
-        mode=None if mode is None else _deferred_overwrite_mode(mode),
-        storage_options=storage_options,
-    )
+    store_path = await make_store_path(store, path=path, mode=mode, storage_options=storage_options)
 
     if "write_empty_chunks" in kwargs:
         _warn_write_empty_chunks_kwarg()
diff --git a/src/zarr/api/synchronous.py b/src/zarr/api/synchronous.py
index 6552b3acb5..58411a0ed9 100644
--- a/src/zarr/api/synchronous.py
+++ b/src/zarr/api/synchronous.py
@@ -200,7 +200,8 @@ def open(
         Persistence mode: 'r' means read only (must exist); 'r+' means
         read/write (must exist); 'a' means read/write (create if doesn't
         exist); 'w' means create (overwrite if exists); 'w-' means create
-        (fail if exists).
+        (fail if exists). On a store that cannot delete keys, 'w' raises an
+        error instead of replacing an existing node.
         If the store is read-only, the default is 'r'; otherwise, it is 'a'.
     zarr_format : {2, 3, None}, optional
         The zarr format to use when saving.
@@ -518,7 +519,8 @@ def open_group(
         Persistence mode: 'r' means read only (must exist); 'r+' means
         read/write (must exist); 'a' means read/write (create if doesn't
         exist); 'w' means create (overwrite if exists); 'w-' means create
-        (fail if exists).
+        (fail if exists). On a store that cannot delete keys, 'w' raises an
+        error instead of replacing an existing node.
     cache_attrs : bool, optional
         If True (default), user attributes will be cached for attribute read
         operations. If False, user attributes are reloaded from the store prior
@@ -800,7 +802,8 @@ def create(
           `True`.
         - `'r'` always fails.
 
-        `mode` has no effect, and is not validated, if `store` is a `StorePath`.
+        If `store` is a `StorePath`, `mode` is not validated against it: `'w'` still
+        sets `overwrite`, and the other modes have no effect.
     data : array-like, optional
         Values written into the new array after it is created. Unlike the `data`
         parameter of `create_array`, it does not set the shape or data type of the
diff --git a/src/zarr/core/metadata/io.py b/src/zarr/core/metadata/io.py
index 834c385a60..afbcb238fc 100644
--- a/src/zarr/core/metadata/io.py
+++ b/src/zarr/core/metadata/io.py
@@ -75,8 +75,9 @@ async def save_new_metadata(
     metadata : ArrayMetadata | GroupMetadata
         Metadata of the new node.
     overwrite : bool
-        If true and the store supports deletes, delete any existing node at `store_path`.
-        Otherwise, raise if a node already exists at `store_path`.
+        If true and the store supports deletes, delete everything under `store_path`.
+        Otherwise, raise if a node already exists at `store_path`; a store that cannot
+        delete keys cannot overwrite a node, so it raises the same error either way.
     ensure_parents : bool, optional
         Create any missing parent groups, and check no existing parents are arrays.
     """
diff --git a/src/zarr/storage/_common.py b/src/zarr/storage/_common.py
index df08968e5c..8591447449 100644
--- a/src/zarr/storage/_common.py
+++ b/src/zarr/storage/_common.py
@@ -75,7 +75,10 @@ async def open(cls, store: Store, path: str, mode: AccessModeLiteral | None = No
         * If the mode is 'r+', 'w-', 'w', or 'a' and the store is read-only, raise a ValueError.
         * If the mode is 'r' and the store is not read-only, return a copy of the store with read_only set to True.
         * If the mode is 'w-' and the store is not read-only and the StorePath contains keys, raise a FileExistsError.
-        * If the mode is 'w'  and the store is not read-only, delete all keys nested within the StorePath.
+
+        No mode modifies the store. In particular, mode 'w' does not delete the keys nested
+        within the StorePath: the node created at the path replaces them, once that node is
+        valid (see `zarr.core.metadata.io.save_new_metadata`).
 
         Parameters
         ----------
@@ -87,7 +90,7 @@ async def open(cls, store: Store, path: str, mode: AccessModeLiteral | None = No
             - `'r'`: read only (must exist)
             - `'r+'`: read/write (must exist)
             - `'a'`: read/write (create if doesn't exist)
-            - `'w'`: read/write (overwrite if exists)
+            - `'w'`: read/write (the node created at the path overwrites what exists)
             - `'w-'`: read/write (create if doesn't exist).
 
         Raises
@@ -126,16 +129,11 @@ async def open(cls, store: Store, path: str, mode: AccessModeLiteral | None = No
             # writable store and writable mode
             self = await cls._create_open_instance(store, path)
 
-        # Handle mode-specific operations
-        match mode:
-            case "w-":
-                if not await self.is_empty():
-                    raise FileExistsError(
-                        f"Cannot create '{path}' with mode 'w-' because it already contains data. "
-                        f"Use mode 'w' to overwrite or 'a' to append."
-                    )
-            case "w":
-                await self.delete_dir()
+        if mode == "w-" and not await self.is_empty():
+            raise FileExistsError(
+                f"Cannot create '{path}' with mode 'w-' because it already contains data. "
+                f"Use mode 'w' to overwrite or 'a' to append."
+            )
         return self
 
     async def get(
diff --git a/tests/test_api.py b/tests/test_api.py
index 54db82de1f..f874bd45b7 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -526,20 +526,38 @@ def test_open_like_zarr_format(source_format: ZarrFormat, target_format: ZarrFor
     }
 
 
-def _store_with_array(zarr_format: ZarrFormat) -> MemoryStore:
-    """A MemoryStore holding a written int32 array of shape (4,) at the root."""
-    store = MemoryStore()
-    arr = zarr.create_array(store, shape=(4,), dtype="int32", zarr_format=zarr_format)
-    arr[:] = 1
-    return store
+async def _read_all(store: Store) -> dict[str, bytes]:
+    contents = {}
+    for key in sorted([key async for key in store.list()]):
+        value = await store.get(key, prototype=default_buffer_prototype())
+        assert value is not None
+        contents[key] = value.to_bytes()
+    return contents
+
+
+def _store_contents(store: Store) -> dict[str, bytes]:
+    """Every key in `store` with its bytes."""
+    return sync(_read_all(store))
 
 
-def _store_contents(store: MemoryStore) -> dict[str, bytes]:
-    return {key: value.to_bytes() for key, value in store._store_dict.items()}
+def _seed_array(store: Store, zarr_format: ZarrFormat) -> None:
+    """Write an int32 array of shape (4,), with its chunk, at the root of `store`."""
+    zarr.create_array(store, shape=(4,), dtype="int32", zarr_format=zarr_format)[:] = 1
 
 
+def _seed_group(store: Store, zarr_format: ZarrFormat) -> None:
+    """Write a group with attributes and a child array, with its chunk, at the root of `store`."""
+    root = zarr.create_group(store, zarr_format=zarr_format, attributes={"k": 1})
+    root.create_array("child", shape=(4,), dtype="int32")[:] = 1
+
+
+_SEEDS: dict[str, Callable[[Store, ZarrFormat], None]] = {
+    "array": _seed_array,
+    "group": _seed_group,
+}
+
 # Each call replaces the node at the root of the store it is given.
-_OVERWRITE_CALLS: dict[str, Callable[..., AnyArray | Group]] = {
+_OVERWRITE_CALLS: dict[str, Callable[..., object]] = {
     "open-array": lambda store, **kw: zarr.open(store=store, mode="w", shape=(2,), **kw),
     "open-group": lambda store, **kw: zarr.open(store=store, mode="w", **kw),
     "open_array": lambda store, **kw: zarr.open_array(store=store, mode="w", shape=(2,), **kw),
@@ -552,26 +570,76 @@ def _store_contents(store: MemoryStore) -> dict[str, bytes]:
         store, overwrite=True, shape=(2,), dtype="int32", **kw
     ),
     "save_array": lambda store, **kw: zarr.save_array(store, np.arange(2), mode="w", **kw),
+    "save_group": lambda store, **kw: zarr.save_group(store, np.arange(2), **kw),
     "group": lambda store, **kw: zarr.group(store=store, overwrite=True, **kw),
 }
 
 
 @pytest.mark.parametrize("zarr_format", [2, 3])
+@pytest.mark.parametrize("existing", _SEEDS)
 @pytest.mark.parametrize("call", _OVERWRITE_CALLS)
-def test_overwrite_replaces_existing_node(zarr_format: ZarrFormat, call: str) -> None:
+def test_overwrite_replaces_existing_node(
+    zarr_format: ZarrFormat, existing: str, call: str
+) -> None:
     """
     Opening or creating a node in an overwriting mode replaces the node at the path,
-    including its chunks.
+    including its chunks or children: the store ends up exactly as the same call leaves
+    an empty store.
     """
-    store = _store_with_array(zarr_format)
+    store = MemoryStore()
+    _SEEDS[existing](store, zarr_format)
     _OVERWRITE_CALLS[call](store)
-    reopened = zarr.open(store=store, mode="r")
-    if call in {"open-group", "open_group", "group"}:
-        assert isinstance(reopened, Group)
-        assert set(store._store_dict) <= {"zarr.json", ".zgroup", ".zattrs"}
-    else:
-        assert isinstance(reopened, Array)
-        assert reopened.shape == (2,)
+    reference = MemoryStore()
+    _OVERWRITE_CALLS[call](reference)
+    assert _store_contents(store) == _store_contents(reference)
+
+
+@pytest.mark.parametrize("call", _OVERWRITE_CALLS)
+def test_overwrite_without_deletes_creates_node(tmp_path: Path, call: str) -> None:
+    """
+    On a store that cannot delete keys, an overwriting call still creates the node when
+    nothing is stored under the path.
+    """
+    with ZipStore(tmp_path / "store.zip", mode="w") as store:
+        _OVERWRITE_CALLS[call](store)
+        reference = MemoryStore()
+        _OVERWRITE_CALLS[call](reference)
+        assert _store_contents(store) == _store_contents(reference)
+
+
+@pytest.mark.parametrize("call", _OVERWRITE_CALLS)
+def test_overwrite_without_deletes_keeps_existing_node(tmp_path: Path, call: str) -> None:
+    """
+    On a store that cannot delete keys, an overwriting call raises without modifying the
+    store when a node already exists at the path.
+    """
+    with ZipStore(tmp_path / "store.zip", mode="w") as store:
+        _seed_array(store, 3)
+        before = _store_contents(store)
+        with pytest.raises(ContainsArrayError):
+            _OVERWRITE_CALLS[call](store)
+        assert _store_contents(store) == before
+
+
+@pytest.mark.parametrize(
+    "call",
+    [
+        "open-array",
+        "open-group",
+        "open_array",
+        "open_group",
+        "create-mode",
+        "save_array",
+        "save_group",
+    ],
+)
+def test_overwrite_read_only_store_names_mode(call: str) -> None:
+    """
+    An overwriting mode on a read-only store is refused with an error that names the mode
+    that was passed.
+    """
+    with pytest.raises(ValueError, match="mode is 'w'"):
+        _OVERWRITE_CALLS[call](MemoryStore(read_only=True))
 
 
 @pytest.mark.parametrize(
@@ -583,7 +651,8 @@ def test_overwrite_invalid_keyword_keeps_existing_node(call: str) -> None:
     An overwriting call that is given a keyword argument its target cannot accept raises
     without modifying the existing node.
     """
-    store = _store_with_array(3)
+    store = MemoryStore()
+    _seed_array(store, 3)
     before = _store_contents(store)
     kwargs = {"config": {"order": "F"}} if call == "open-group" else {"bogus": 1}
     with pytest.raises(TypeError, match="unexpected keyword argument"):
@@ -600,7 +669,8 @@ def test_overwrite_invalid_fill_value_keeps_existing_node(call: str) -> None:
     An overwriting call whose new array metadata cannot be built raises without modifying
     the existing node.
     """
-    store = _store_with_array(3)
+    store = MemoryStore()
+    _seed_array(store, 3)
     before = _store_contents(store)
     kwargs: dict[str, Any] = {"fill_value": "not a number"}
     if call in {"open-array", "open_array", "create-mode", "create-overwrite"}:
@@ -619,25 +689,40 @@ def test_overwrite_unencodable_attributes_keeps_existing_node(call: str) -> None
     An overwriting call whose new metadata cannot be encoded raises without modifying the
     existing node.
     """
-    store = _store_with_array(3)
+    store = MemoryStore()
+    _seed_array(store, 3)
     before = _store_contents(store)
     with pytest.raises(TypeError, match="not JSON serializable"):
         _OVERWRITE_CALLS[call](store, attributes={"x": object()})
     assert _store_contents(store) == before
 
 
-def test_save_group_invalid_argument_keeps_existing_node() -> None:
+def test_save_group_non_array_argument_keeps_existing_node() -> None:
     """
     save_group given an argument that is not an array raises without modifying the
     existing node.
     """
-    store = _store_with_array(3)
+    store = MemoryStore()
+    _seed_group(store, 3)
     before = _store_contents(store)
     with pytest.raises(TypeError, match="must be a numpy or other NDArrayLike array"):
         zarr.save_group(store, a=np.arange(2), b="not an array")  # type: ignore[arg-type]
     assert _store_contents(store) == before
 
 
+def test_save_group_unstorable_dtype_keeps_existing_node() -> None:
+    """
+    save_group given an array whose data type Zarr cannot store raises without modifying
+    the existing node.
+    """
+    store = MemoryStore()
+    _seed_group(store, 3)
+    before = _store_contents(store)
+    with pytest.raises(ValueError, match="data type resolution"):
+        zarr.save_group(store, a=np.arange(2), b=np.array([object()], dtype=object))
+    assert _store_contents(store) == before
+
+
 @pytest.mark.parametrize("store", ["memory", "local", "zip"], indirect=True)
 def test_v2_and_v3_exist_at_same_path(store: Store) -> None:
     zarr.create_array(store, shape=(10,), dtype="uint8", zarr_format=3)
diff --git a/tests/test_store/test_core.py b/tests/test_store/test_core.py
index f39dd0cfd3..daf62e3843 100644
--- a/tests/test_store/test_core.py
+++ b/tests/test_store/test_core.py
@@ -230,6 +230,17 @@ async def test_store_path_invalid_mode_raises(
         await StorePath.open(LocalStore(str(tmp_path), read_only=modes[0]), path="", mode=modes[1])  # type: ignore[arg-type]
 
 
+async def test_store_path_open_w_keeps_keys(tmp_path: Path) -> None:
+    """
+    Opening a store path with mode 'w' leaves the keys under it in place: the node created
+    at the path replaces them once it is valid, not the act of opening the path.
+    """
+    store = LocalStore(str(tmp_path))
+    await (StorePath(store, "root") / "key").set(cpu.Buffer.from_bytes(b"data"))
+    store_path = await StorePath.open(store, path="root", mode="w")
+    assert not await store_path.is_empty()
+
+
 async def test_make_store_invalid_mode_raises() -> None:
     """
     Test that make_store raises ValueError for a mode outside the access-mode literals.