diff --git a/changes/4366.misc.md b/changes/4366.misc.md
new file mode 100644
index 0000000000..b04feda153
--- /dev/null
+++ b/changes/4366.misc.md
@@ -0,0 +1 @@
+`zarr.open`, `open_array`, `open_group`, `AsyncArray.open` and `AsyncGroup.open` are now built on one metadata reader, `zarr.core.group.read_node_metadata`, composed of a per-format `read_v3_metadata` and `read_v2_metadata`; each reads a node's documents once and returns the parsed metadata whichever kind the node is, and with `zarr_format=None` format 3 is tried first and the format 2 documents are read only when there is no `zarr.json`. Opening a Zarr format 3 node costs a single read where `zarr.open` used to make up to seven. `zarr.open` now has a stated contract: with `shape` it behaves as `open_array`; otherwise the reading modes open whatever node is there ('a' creating a group when there is none) and the creating modes create a group ('w' replacing whatever is there). Consequences: `zarr.open(mode="w")` replaces an existing array with a group instead of returning it; `zarr.open(shape=...)` on a group raises `ContainsGroupError` instead of opening the group; `zarr.open(mode="r")` on nothing raises `NodeNotFoundError`; opening an array as a group raises `ContainsArrayError` and a group as an array `ContainsGroupError` in every format and mode; a Zarr format 3 document whose `node_type` is missing or unknown raises `NodeTypeValidationError` everywhere; and a path holding both formats is read as format 3 without a warning.
diff --git a/src/zarr/api/asynchronous.py b/src/zarr/api/asynchronous.py
index 1fc10cdd1e..f29a891c8e 100644
--- a/src/zarr/api/asynchronous.py
+++ b/src/zarr/api/asynchronous.py
@@ -3,7 +3,7 @@
 import asyncio
 import dataclasses
 import warnings
-from typing import TYPE_CHECKING, Any, Literal, NotRequired, TypedDict, cast
+from typing import TYPE_CHECKING, Any, Literal, NotRequired, TypedDict
 
 import numpy as np
 import numpy.typing as npt
@@ -17,7 +17,6 @@
     CompressorLike,
     create_array,
     from_array,
-    get_array_metadata,
 )
 from zarr.core.array_spec import ArrayConfigLike, parse_array_config
 from zarr.core.buffer import NDArrayLike
@@ -35,13 +34,16 @@
     AsyncGroup,
     ConsolidatedMetadata,
     GroupMetadata,
+    _open_node,
     create_hierarchy,
 )
-from zarr.core.metadata import ArrayMetadataDict, ArrayV2Metadata
+from zarr.core.metadata import ArrayV2Metadata
 from zarr.errors import (
     ArrayNotFoundError,
+    ContainsArrayError,
+    ContainsGroupError,
     GroupNotFoundError,
-    NodeTypeValidationError,
+    NodeNotFoundError,
     ZarrDeprecationWarning,
     ZarrRuntimeWarning,
     ZarrUserWarning,
@@ -356,7 +358,8 @@ async def open(
         (fail if exists).
         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.
+        The Zarr format of the node. None opens whichever format is found,
+        trying Zarr format 3 first, and creates the default format.
     path : str or None, optional
         The path within the store to open.
     storage_options : dict
@@ -377,6 +380,15 @@ async def open(
 
     Notes
     -----
+    What `open` opens or creates follows two rules. If `shape` is given, the
+    call describes an array and behaves as
+    [`open_array`][zarr.api.asynchronous.open_array] with the same arguments.
+    Otherwise, in the modes that read ('r', 'r+' and 'a'), the node at `path` is
+    opened whichever kind it is; when there is none, 'r' and 'r+' raise
+    [`NodeNotFoundError`][zarr.errors.NodeNotFoundError] and 'a' creates a
+    group. The modes that only create ('w' and 'w-') create a group, 'w'
+    replacing whatever is at `path` and 'w-' failing if anything is.
+
     `open` returns a lazy [`Array`][zarr.Array] or [`Group`][zarr.Group] backed by
     the store, so data is read and written incrementally. Use [`load`][zarr.load]
     instead when you want the data eagerly read into an in-memory array (a
@@ -390,30 +402,23 @@ async def open(
             mode = "a"
     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"}:
-        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:
+    if "shape" in kwargs:
+        # the call describes an array
         return await open_array(store=store_path, zarr_format=zarr_format, mode=mode, **kwargs)
-    except (KeyError, NodeTypeValidationError):
-        # KeyError for a missing key
-        # NodeTypeValidationError for failing to parse node metadata as an array when it's
-        # actually a group
-        return await open_group(store=store_path, zarr_format=zarr_format, mode=mode, **kwargs)
+    if mode in _READ_MODES:
+        node = await _open_node(
+            store_path,
+            zarr_format=zarr_format,
+            use_consolidated=kwargs.get("use_consolidated"),
+            config=kwargs.get("config"),
+        )
+        if node is not None:
+            return node
+        if mode != "a":
+            msg = f"No array or group found in store {store_path.store} at path {store_path.path!r}"
+            raise NodeNotFoundError(msg)
+    # nothing to open, or a mode that only creates: make a group
+    return await open_group(store=store_path, zarr_format=zarr_format, mode=mode, **kwargs)
 
 
 async def open_consolidated(
@@ -857,24 +862,21 @@ async def open_group(
     )
 
     store_path = await make_store_path(store, mode=mode, storage_options=storage_options, path=path)
-    if attributes is None:
-        attributes = {}
-
-    try:
-        if mode in _READ_MODES:
-            return await AsyncGroup.open(
-                store_path, zarr_format=zarr_format, use_consolidated=use_consolidated
-            )
-    except (KeyError, FileNotFoundError):
-        pass
+    if mode in _READ_MODES:
+        node = await _open_node(
+            store_path, zarr_format=zarr_format, use_consolidated=use_consolidated
+        )
+        if isinstance(node, AsyncGroup):
+            return node
+        if node is not None:
+            msg = f"An array exists in store {store_path.store} at path {store_path.path}."
+            raise ContainsArrayError(msg)
     if mode in _CREATE_MODES:
-        overwrite = _infer_overwrite(mode)
-        _zarr_format = zarr_format or _default_zarr_format()
         return await AsyncGroup.from_store(
             store_path,
-            zarr_format=_zarr_format,
-            overwrite=overwrite,
-            attributes=attributes,
+            zarr_format=zarr_format or _default_zarr_format(),
+            overwrite=_infer_overwrite(mode),
+            attributes=attributes or {},
         )
     msg = f"No group found in store {store!r} at path {store_path.path!r}"
     raise GroupNotFoundError(msg)
@@ -1266,20 +1268,26 @@ async def open_array(
     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
+    if mode not in _OVERWRITE_MODES:
+        # Whatever is here is what the caller gets, unless it is a group. An array
+        # has no consolidated metadata to read.
+        node = await _open_node(
+            store_path, zarr_format=zarr_format, use_consolidated=False, config=kwargs.get("config")
+        )
+        if isinstance(node, AsyncArray):
+            return node
+        if node is not None:
+            msg = f"A group exists in store {store_path.store} at path {store_path.path}."
+            raise ContainsGroupError(msg)
+    if not store_path.read_only and mode in _CREATE_MODES:
+        return await create(
+            store=store_path,
+            zarr_format=zarr_format or _default_zarr_format(),
+            overwrite=_infer_overwrite(mode),
+            **kwargs,
+        )
+    msg = f"No array found in store {store_path.store} at path {store_path.path}"
+    raise ArrayNotFoundError(msg)
 
 
 async def open_like(a: ArrayLike, path: str, **kwargs: Any) -> AnyAsyncArray:
diff --git a/src/zarr/api/synchronous.py b/src/zarr/api/synchronous.py
index 6975f6d953..595345deac 100644
--- a/src/zarr/api/synchronous.py
+++ b/src/zarr/api/synchronous.py
@@ -203,7 +203,8 @@ def open(
         (fail if exists).
         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.
+        The Zarr format of the node. None opens whichever format is found,
+        trying Zarr format 3 first, and creates the default format.
     path : str or None, optional
         The path within the store to open.
     storage_options : dict
@@ -224,6 +225,14 @@ def open(
 
     Notes
     -----
+    What `open` opens or creates follows two rules. If `shape` is given, the
+    call describes an array and behaves as [`open_array`][zarr.open_array] with
+    the same arguments. Otherwise, in the modes that read ('r', 'r+' and 'a'),
+    the node at `path` is opened whichever kind it is; when there is none, 'r'
+    and 'r+' raise [`NodeNotFoundError`][zarr.errors.NodeNotFoundError] and 'a'
+    creates a group. The modes that only create ('w' and 'w-') create a group,
+    'w' replacing whatever is at `path` and 'w-' failing if anything is.
+
     `open` returns a lazy [`Array`][zarr.Array] or [`Group`][zarr.Group] backed by
     the store, so data is read and written incrementally. Use [`load`][zarr.load]
     instead when you want the data eagerly read into an in-memory array (a
diff --git a/src/zarr/core/array.py b/src/zarr/core/array.py
index 5a8d6bf57e..480c526681 100644
--- a/src/zarr/core/array.py
+++ b/src/zarr/core/array.py
@@ -3,7 +3,6 @@
 import copy
 import math
 import warnings
-from asyncio import gather
 from collections.abc import Iterable, Mapping, Sequence
 from dataclasses import dataclass, field, replace
 from itertools import starmap
@@ -29,7 +28,6 @@
 from zarr.codecs.vlen_utf8 import VLenBytesCodec, VLenUTF8Codec
 from zarr.codecs.zstd import ZstdCodec
 from zarr.core._info import ArrayInfo
-from zarr.core._json import buffer_to_json_object
 from zarr.core.array_spec import ArrayConfig, ArrayConfigLike, ArraySpec, parse_array_config
 from zarr.core.attributes import Attributes
 from zarr.core.buffer import (
@@ -39,7 +37,6 @@
     NDBuffer,
     default_buffer_prototype,
 )
-from zarr.core.buffer.cpu import buffer_prototype as cpu_buffer_prototype
 from zarr.core.chunk_grids import (
     SHARDED_INNER_CHUNK_MAX_BYTES,
     ChunkGrid,
@@ -59,9 +56,6 @@
 )
 from zarr.core.common import (
     JSON,
-    ZARR_JSON,
-    ZARRAY_JSON,
-    ZATTRS_JSON,
     ChunksLike,
     DimensionNamesLike,
     MemoryOrder,
@@ -130,13 +124,12 @@
     RectilinearChunkGridMetadata,
     RegularChunkGridMetadata,
     create_chunk_grid_metadata,
-    parse_node_type_array,
 )
 from zarr.core.sync import sync
 from zarr.errors import (
     ArrayNotFoundError,
     ChunkNotFoundError,
-    MetadataValidationError,
+    ContainsGroupError,
     ZarrDeprecationWarning,
     ZarrUserWarning,
 )
@@ -158,7 +151,6 @@
     from zarr.abc.codec import CodecPipeline
     from zarr.abc.store import Store
     from zarr.codecs.sharding import IndexLocation, ShardingCodec
-    from zarr.core.buffer import Buffer
     from zarr.core.dtype.wrapper import TBaseDType, TBaseScalar
     from zarr.storage import StoreLike
     from zarr.types import AnyArray, AnyAsyncArray, ArrayV2, ArrayV3, AsyncArrayV2, AsyncArrayV3
@@ -269,69 +261,36 @@ def create_codec_pipeline(metadata: ArrayMetadata, *, store: Store | None = None
     raise TypeError  # pragma: no cover
 
 
-async def get_array_metadata(
-    store_path: StorePath, zarr_format: ZarrFormat | None = 3
-) -> dict[str, JSON]:
-    if zarr_format == 2:
-        zarray_bytes, zattrs_bytes = await gather(
-            (store_path / ZARRAY_JSON).get(prototype=cpu_buffer_prototype),
-            (store_path / ZATTRS_JSON).get(prototype=cpu_buffer_prototype),
-        )
-        if zarray_bytes is None:
-            msg = (
-                "A Zarr V2 array metadata document was not found in store "
-                f"{store_path.store!r} at path {store_path.path!r}."
-            )
-            raise ArrayNotFoundError(msg)
-        return _array_metadata_dict_v2(zarray_bytes, zattrs_bytes)
-    elif zarr_format == 3:
-        zarr_json_bytes = await (store_path / ZARR_JSON).get(prototype=cpu_buffer_prototype)
-        if zarr_json_bytes is None:
-            msg = (
-                "A Zarr V3 array metadata document was not found in store "
-                f"{store_path.store!r} at path {store_path.path!r}."
-            )
-            raise ArrayNotFoundError(msg)
-        return _array_metadata_dict_v3(zarr_json_bytes)
-    elif zarr_format is None:
-        zarr_json_bytes, zarray_bytes, zattrs_bytes = await gather(
-            (store_path / ZARR_JSON).get(prototype=cpu_buffer_prototype),
-            (store_path / ZARRAY_JSON).get(prototype=cpu_buffer_prototype),
-            (store_path / ZATTRS_JSON).get(prototype=cpu_buffer_prototype),
-        )
-        if zarr_json_bytes is not None and zarray_bytes is not None:
-            # warn and favor v3
-            msg = f"Both zarr.json (Zarr format 3) and .zarray (Zarr format 2) metadata objects exist at {store_path}. Zarr v3 will be used."
-            warnings.warn(msg, category=ZarrUserWarning, stacklevel=1)
-        # favor v3 when both are present
-        if zarr_json_bytes is not None:
-            return _array_metadata_dict_v3(zarr_json_bytes)
-        if zarray_bytes is not None:
-            return _array_metadata_dict_v2(zarray_bytes, zattrs_bytes)
-        msg = (
-            f"Neither Zarr V3 nor Zarr V2 array metadata documents "
-            f"were found in store {store_path.store!r} at path {store_path.path!r}."
-        )
-        raise ArrayNotFoundError(msg)
-    else:
-        msg = f"Invalid value for 'zarr_format'. Expected 2, 3, or None. Got '{zarr_format}'."  # type: ignore[unreachable]
-        raise MetadataValidationError(msg)
+async def _read_array_metadata(
+    store_path: StorePath, zarr_format: ZarrFormat | None
+) -> ArrayMetadata:
+    """The array metadata at `store_path`.
 
+    Raises `ArrayNotFoundError` if there is no node there and
+    `ContainsGroupError` if the node is a group.
+    """
+    # group.py imports this module, so the shared reader is imported here
+    from zarr.core.group import GroupMetadata, read_node_metadata
 
-def _array_metadata_dict_v2(zarray_bytes: Buffer, zattrs_bytes: Buffer | None) -> dict[str, JSON]:
-    """Combine a `.zarray` document and an optional `.zattrs` document into one metadata dict."""
-    metadata_dict: dict[str, JSON] = buffer_to_json_object(zarray_bytes)
-    metadata_dict["attributes"] = (
-        buffer_to_json_object(zattrs_bytes) if zattrs_bytes is not None else {}
-    )
-    return metadata_dict
+    metadata = await read_node_metadata(store_path.store, store_path.path, zarr_format)
+    if metadata is None:
+        msg = f"No array found in store {store_path.store} at path {store_path.path!r}"
+        raise ArrayNotFoundError(msg)
+    if isinstance(metadata, GroupMetadata):
+        msg = f"A group exists in store {store_path.store} at path {store_path.path}."
+        raise ContainsGroupError(msg)
+    return metadata
 
 
-def _array_metadata_dict_v3(zarr_json_bytes: Buffer) -> dict[str, JSON]:
-    """Parse a `zarr.json` document, checking that it describes an array."""
-    metadata_dict: dict[str, JSON] = buffer_to_json_object(zarr_json_bytes)
-    parse_node_type_array(metadata_dict.get("node_type"))
-    return metadata_dict
+async def get_array_metadata(
+    store_path: StorePath, zarr_format: ZarrFormat | None = 3
+) -> dict[str, JSON]:
+    """The array metadata document at `store_path`, as a dict.
+
+    Prefer `AsyncArray.open`, which builds the array from the parsed metadata
+    directly; this is kept for callers that want the document.
+    """
+    return (await _read_array_metadata(store_path, zarr_format)).to_dict()
 
 
 async def _prepare_overwrite(
@@ -813,10 +772,8 @@ async def example():
         ```
         """
         store_path = await make_store_path(store)
-        metadata_dict = await get_array_metadata(store_path, zarr_format=zarr_format)
-        # TODO: remove this cast when we have better type hints
-        _metadata_dict = cast("ArrayMetadataJSON_V3", metadata_dict)
-        return cls(store_path=store_path, metadata=_metadata_dict)
+        metadata = await _read_array_metadata(store_path, zarr_format)
+        return cls(store_path=store_path, metadata=metadata)
 
     @property
     def store(self) -> Store:
diff --git a/src/zarr/core/group.py b/src/zarr/core/group.py
index d734e6b7cd..46a5555cc7 100644
--- a/src/zarr/core/group.py
+++ b/src/zarr/core/group.py
@@ -85,6 +85,83 @@
 logger = logging.getLogger("zarr.group")
 
 
+def _resolve_use_consolidated(
+    store: Store, use_consolidated: bool | str | None
+) -> bool | str | None:
+    """Settle `use_consolidated` against what `store` supports.
+
+    A store that can't hold consolidated metadata makes the answer False, unless
+    consolidated metadata was explicitly asked for, which is an error.
+    """
+    if store.supports_consolidated_metadata:
+        return use_consolidated
+    if use_consolidated:
+        raise ValueError(
+            f"The Zarr store in use ({type(store).__name__}) doesn't support consolidated metadata."
+        )
+    return False
+
+
+def _v2_consolidated_key(use_consolidated: bool | str | None) -> str | None:
+    """The Zarr format 2 consolidated-metadata key to read for `use_consolidated`, or None for none."""
+    if use_consolidated is False:
+        return None
+    if isinstance(use_consolidated, str):
+        return use_consolidated
+    return ZMETADATA_V2_JSON
+
+
+def _apply_use_consolidated(
+    metadata: GroupMetadata, use_consolidated: bool | str | None, *, store_path: StorePath
+) -> GroupMetadata:
+    """Enforce the caller's `use_consolidated` on freshly read group metadata.
+
+    True (or a format 2 key) requires consolidated metadata to be present; False
+    drops whatever is present; None keeps whatever is present.
+    """
+    if metadata.zarr_format == 3 and not isinstance(use_consolidated, bool | None):
+        raise TypeError("use_consolidated must be a bool or None for Zarr format 3.")
+    if use_consolidated and metadata.consolidated_metadata is None:
+        msg = (
+            f"Consolidated metadata requested with 'use_consolidated={use_consolidated!r}' "
+            f"but not found in '{store_path.path}'."
+        )
+        raise ValueError(msg)
+    if use_consolidated is False and metadata.consolidated_metadata is not None:
+        return replace(metadata, consolidated_metadata=None)
+    return metadata
+
+
+async def _open_node(
+    store_path: StorePath,
+    *,
+    zarr_format: ZarrFormat | None,
+    use_consolidated: bool | str | None,
+    config: ArrayConfigLike | None = None,
+) -> AnyAsyncArray | AsyncGroup | None:
+    """The node at `store_path`, whichever kind it is, or None if there is none.
+
+    This is what the opening functions are built from: it reads the node's
+    metadata once, applies `use_consolidated` if the node is a group, and
+    builds the node. It applies no mode policy and raises nothing for a missing
+    node; the callers decide what those mean.
+    """
+    use_consolidated = _resolve_use_consolidated(store_path.store, use_consolidated)
+    metadata = await read_node_metadata(
+        store_path.store,
+        store_path.path,
+        zarr_format,
+        consolidated_key=_v2_consolidated_key(use_consolidated),
+    )
+    if metadata is None:
+        return None
+    if isinstance(metadata, GroupMetadata):
+        metadata = _apply_use_consolidated(metadata, use_consolidated, store_path=store_path)
+    return _build_node(
+        store=store_path.store, path=store_path.path, metadata=metadata, config=config
+    )
+
+
 def parse_zarr_format(data: Any) -> ZarrFormat:
     """Parse the zarr_format field from metadata."""
     return cast("ZarrFormat", parse_field(data, Literal[2, 3], "zarr_format"))
@@ -525,161 +602,16 @@ async def open(
             to load consolidated metadata from a non-default key.
         """
         store_path = await make_store_path(store)
-        if not store_path.store.supports_consolidated_metadata:
-            # Fail if consolidated metadata was requested but the Store doesn't support it
-            if use_consolidated:
-                store_name = type(store_path.store).__name__
-                raise ValueError(
-                    f"The Zarr store in use ({store_name}) doesn't support consolidated metadata."
-                )
-
-            # if use_consolidated was None (optional), the Store dictates it doesn't want consolidation
-            use_consolidated = False
-
-        consolidated_key = ZMETADATA_V2_JSON
-
-        if (zarr_format == 2 or zarr_format is None) and isinstance(use_consolidated, str):
-            consolidated_key = use_consolidated
-
-        if zarr_format == 2:
-            paths = [store_path / ZGROUP_JSON, store_path / ZATTRS_JSON]
-            if use_consolidated or use_consolidated is None:
-                paths.append(store_path / consolidated_key)
-
-            zgroup_bytes, zattrs_bytes, *rest = await asyncio.gather(
-                *[path.get() for path in paths]
-            )
-            if zgroup_bytes is None:
-                raise FileNotFoundError(store_path)
-
-            if use_consolidated or use_consolidated is None:
-                maybe_consolidated_metadata_bytes = rest[0]
-
-            else:
-                maybe_consolidated_metadata_bytes = None
-
-        elif zarr_format == 3:
-            zarr_json_bytes = await (store_path / ZARR_JSON).get()
-            if zarr_json_bytes is None:
-                raise FileNotFoundError(store_path)
-        elif zarr_format is None:
-            (
-                zarr_json_bytes,
-                zgroup_bytes,
-                zattrs_bytes,
-                maybe_consolidated_metadata_bytes,
-            ) = await asyncio.gather(
-                (store_path / ZARR_JSON).get(),
-                (store_path / ZGROUP_JSON).get(),
-                (store_path / ZATTRS_JSON).get(),
-                (store_path / str(consolidated_key)).get(),
-            )
-            if zarr_json_bytes is not None and zgroup_bytes is not None:
-                # warn and favor v3
-                msg = f"Both zarr.json (Zarr format 3) and .zgroup (Zarr format 2) metadata objects exist at {store_path}. Zarr format 3 will be used."
-                warnings.warn(msg, category=ZarrUserWarning, stacklevel=1)
-            if zarr_json_bytes is None and zgroup_bytes is None:
-                raise FileNotFoundError(
-                    f"could not find zarr.json or .zgroup objects in {store_path}"
-                )
-            # set zarr_format based on which keys were found
-            if zarr_json_bytes is not None:
-                zarr_format = 3
-            else:
-                zarr_format = 2
-        else:
-            msg = f"Invalid value for 'zarr_format'. Expected 2, 3, or None. Got '{zarr_format}'."  # type: ignore[unreachable]
-            raise MetadataValidationError(msg)
-
-        if zarr_format == 2:
-            if zgroup_bytes is None:
-                raise FileNotFoundError(store_path)
-
-            if use_consolidated and maybe_consolidated_metadata_bytes is None:
-                # the user requested consolidated metadata, but it was missing
-                raise ValueError(consolidated_key)
-
-            elif use_consolidated is False:
-                # the user explicitly opted out of consolidated_metadata.
-                # Discard anything we might have read.
-                maybe_consolidated_metadata_bytes = None
-
-            return cls._from_bytes_v2(
-                store_path, zgroup_bytes, zattrs_bytes, maybe_consolidated_metadata_bytes
-            )
-        else:
-            # V3 groups are comprised of a zarr.json object
-            if zarr_json_bytes is None:
-                raise FileNotFoundError(store_path)
-            if not isinstance(use_consolidated, bool | None):
-                raise TypeError("use_consolidated must be a bool or None for Zarr format 3.")
-
-            return cls._from_bytes_v3(
-                store_path,
-                zarr_json_bytes,
-                use_consolidated=use_consolidated,
-            )
-
-    @classmethod
-    def _from_bytes_v2(
-        cls,
-        store_path: StorePath,
-        zgroup_bytes: Buffer,
-        zattrs_bytes: Buffer | None,
-        consolidated_metadata_bytes: Buffer | None,
-    ) -> AsyncGroup:
-        # V2 groups are comprised of a .zgroup and .zattrs objects
-        zgroup = buffer_to_json_object(zgroup_bytes)
-        zattrs = buffer_to_json_object(zattrs_bytes) if zattrs_bytes is not None else {}
-        group_metadata: dict[str, Any] = {**zgroup, "attributes": zattrs}
-
-        if consolidated_metadata_bytes is not None:
-            v2_consolidated_doc = buffer_to_json_object(consolidated_metadata_bytes)
-            v2_consolidated_metadata = cast("dict[str, Any]", v2_consolidated_doc["metadata"])
-            # We already read zattrs and zgroup. Should we ignore these?
-            v2_consolidated_metadata.pop(".zattrs", None)
-            v2_consolidated_metadata.pop(".zgroup", None)
-
-            consolidated_metadata: defaultdict[str, dict[str, Any]] = defaultdict(dict)
-
-            # keys like air/.zarray, air/.zattrs
-            for k, v in v2_consolidated_metadata.items():
-                path, kind = k.rsplit("/.", 1)
-
-                if kind == "zarray":
-                    consolidated_metadata[path].update(v)
-                elif kind == "zattrs":
-                    consolidated_metadata[path]["attributes"] = v
-                elif kind == "zgroup":
-                    consolidated_metadata[path].update(v)
-                else:
-                    raise ValueError(f"Invalid file type '{kind}' at path '{path}")
-
-            group_metadata["consolidated_metadata"] = {
-                "metadata": dict(consolidated_metadata),
-                "kind": "inline",
-                "must_understand": False,
-            }
-
-        return cls.from_dict(store_path, group_metadata)
-
-    @classmethod
-    def _from_bytes_v3(
-        cls,
-        store_path: StorePath,
-        zarr_json_bytes: Buffer,
-        use_consolidated: bool | None,
-    ) -> AsyncGroup:
-        group_metadata = buffer_to_json_object(zarr_json_bytes)
-        if use_consolidated and group_metadata.get("consolidated_metadata") is None:
-            msg = f"Consolidated metadata requested with 'use_consolidated=True' but not found in '{store_path.path}'."
-            raise ValueError(msg)
-
-        elif use_consolidated is False:
-            # Drop consolidated metadata if it's there.
-            group_metadata.pop("consolidated_metadata", None)
-
-        return cls.from_dict(store_path, group_metadata)
+        node = await _open_node(
+            store_path, zarr_format=zarr_format, use_consolidated=use_consolidated
+        )
+        if node is None:
+            msg = f"No group found in store {store_path.store} at path {store_path.path!r}"
+            raise GroupNotFoundError(msg)
+        if not isinstance(node, AsyncGroup):
+            msg = f"An array exists in store {store_path.store} at path {store_path.path}."
+            raise ContainsArrayError(msg)
+        return node
 
     @classmethod
     def from_dict(
@@ -687,13 +619,10 @@ def from_dict(
         store_path: StorePath,
         data: dict[str, Any],
     ) -> AsyncGroup:
-        node_type = data.pop("node_type", None)
-        if node_type == "array":
+        if data.get("node_type") == "array":
             msg = f"An array already exists in store {store_path.store} at path {store_path.path}."
             raise ContainsArrayError(msg)
-        elif node_type not in ("group", None):
-            msg = f"Node type in metadata ({node_type}) is not 'group'"
-            raise GroupNotFoundError(msg)
+        # any other wrong node_type is GroupMetadata.from_dict's to reject
         return cls(
             metadata=GroupMetadata.from_dict(data),
             store_path=store_path,
@@ -3493,53 +3422,134 @@ async def _iter_members_deep(
             yield key, node
 
 
-async def _read_metadata_v3(store: Store, path: str) -> ArrayV3Metadata | GroupMetadata:
+@overload
+async def read_node_metadata(
+    store: Store, path: str, zarr_format: Literal[3], *, consolidated_key: str | None = None
+) -> ArrayV3Metadata | GroupMetadata | None: ...
+
+
+@overload
+async def read_node_metadata(
+    store: Store, path: str, zarr_format: Literal[2], *, consolidated_key: str | None = None
+) -> ArrayV2Metadata | GroupMetadata | None: ...
+
+
+@overload
+async def read_node_metadata(
+    store: Store, path: str, zarr_format: ZarrFormat | None, *, consolidated_key: str | None = None
+) -> ArrayV2Metadata | ArrayV3Metadata | GroupMetadata | None: ...
+
+
+async def read_node_metadata(
+    store: Store, path: str, zarr_format: ZarrFormat | None, *, consolidated_key: str | None = None
+) -> ArrayV2Metadata | ArrayV3Metadata | GroupMetadata | None:
+    """Read the metadata of the node at `path`, whichever kind and format it is.
+
+    With `zarr_format` None, Zarr format 3 is tried first and format 2 only if
+    there is no `zarr.json`, so a path holding both formats is read as format 3
+    without a second look. `consolidated_key` is passed to the format 2 reader.
+    Returns None if no node is found.
     """
-    Given a store_path, return ArrayV3Metadata or GroupMetadata defined by the metadata
-    document stored at store_path.path / zarr.json. If no such document is found, raise a
-    FileNotFoundError.
+    if zarr_format == 3:
+        return await read_v3_metadata(store, path)
+    if zarr_format == 2:
+        return await read_v2_metadata(store, path, consolidated_key=consolidated_key)
+    if zarr_format is None:
+        metadata = await read_v3_metadata(store, path)
+        if metadata is None:
+            return await read_v2_metadata(store, path, consolidated_key=consolidated_key)
+        return metadata
+    msg = f"Invalid value for 'zarr_format'. Expected 2, 3, or None. Got '{zarr_format}'."  # type: ignore[unreachable]
+    raise MetadataValidationError(msg)
+
+
+async def read_v3_metadata(store: Store, path: str) -> ArrayV3Metadata | GroupMetadata | None:
+    """Read the Zarr format 3 node metadata at `path`, or None if there is no `zarr.json`.
+
+    One read: the document's `node_type` says whether it is an array or a group.
     """
     zarr_json_bytes = await store.get(
         _join_paths([path, ZARR_JSON]), prototype=default_buffer_prototype()
     )
     if zarr_json_bytes is None:
-        raise FileNotFoundError(path)
+        return None
     return _build_metadata_v3(buffer_to_json_object(zarr_json_bytes))
 
 
-async def _read_metadata_v2(store: Store, path: str) -> ArrayV2Metadata | GroupMetadata:
-    """
-    Given a store_path, return ArrayV2Metadata or GroupMetadata defined by the metadata
-    document stored at store_path.path / (.zgroup | .zarray). If no such document is found,
-    raise a FileNotFoundError.
+async def read_v2_metadata(
+    store: Store, path: str, *, consolidated_key: str | None = None
+) -> ArrayV2Metadata | GroupMetadata | None:
+    """Read the Zarr format 2 node metadata at `path`, or None if there is neither `.zarray` nor `.zgroup`.
+
+    One concurrent read of `.zarray`, `.zgroup`, `.zattrs` and, if given, the
+    consolidated-metadata document at `consolidated_key`. `.zarray` makes the
+    node an array and `.zgroup` a group, the array winning if both exist; a
+    consolidated document is attached to a group's metadata.
     """
-    # TODO: consider first fetching array metadata, and only fetching group metadata when we don't
-    # find an array
-    zarray_bytes, zgroup_bytes, zattrs_bytes = await asyncio.gather(
-        store.get(_join_paths([path, ZARRAY_JSON]), prototype=default_buffer_prototype()),
-        store.get(_join_paths([path, ZGROUP_JSON]), prototype=default_buffer_prototype()),
-        store.get(_join_paths([path, ZATTRS_JSON]), prototype=default_buffer_prototype()),
+    keys = [ZARRAY_JSON, ZGROUP_JSON, ZATTRS_JSON]
+    if consolidated_key is not None:
+        keys.append(consolidated_key)
+    zarray_bytes, zgroup_bytes, zattrs_bytes, *rest = await asyncio.gather(
+        *(store.get(_join_paths([path, key]), prototype=default_buffer_prototype()) for key in keys)
     )
+    zattrs: dict[str, JSON] = {} if zattrs_bytes is None else buffer_to_json_object(zattrs_bytes)
+    if zarray_bytes is not None:
+        return _build_metadata_v2(buffer_to_json_object(zarray_bytes), zattrs)
+    if zgroup_bytes is None:
+        return None
+    metadata = _build_metadata_v2(buffer_to_json_object(zgroup_bytes), zattrs)
+    consolidated_bytes = rest[0] if rest else None
+    if consolidated_bytes is not None and isinstance(metadata, GroupMetadata):
+        consolidated = _consolidated_metadata_from_v2_doc(buffer_to_json_object(consolidated_bytes))
+        metadata = replace(metadata, consolidated_metadata=consolidated)
+    return metadata
 
-    zattrs: dict[str, JSON]
-    if zattrs_bytes is None:
-        zattrs = {}
-    else:
-        zattrs = buffer_to_json_object(zattrs_bytes)
 
-    # TODO: decide how to handle finding both array and group metadata. The spec does not seem to
-    # consider this situation. A practical approach would be to ignore that combination, and only
-    # return the array metadata.
-    if zarray_bytes is not None:
-        zmeta = buffer_to_json_object(zarray_bytes)
-    else:
-        if zgroup_bytes is None:
-            # neither .zarray or .zgroup were found results in KeyError
-            raise FileNotFoundError(path)
+def _consolidated_metadata_from_v2_doc(doc: dict[str, JSON]) -> ConsolidatedMetadata:
+    """Turn a Zarr format 2 consolidated-metadata document into `ConsolidatedMetadata`.
+
+    The format 2 document is flat, keyed like `air/.zarray` and `air/.zattrs`;
+    those become one metadata dict per path. The root's own `.zgroup` and
+    `.zattrs` are dropped, since they were read directly.
+    """
+    v2_metadata = doc.get("metadata")
+    if not isinstance(v2_metadata, dict):
+        msg = f"A consolidated metadata document needs a 'metadata' object. Got {doc!r}."
+        raise MetadataValidationError(msg)
+    v2_metadata = cast("dict[str, dict[str, Any]]", dict(v2_metadata))
+    v2_metadata.pop(".zattrs", None)
+    v2_metadata.pop(".zgroup", None)
+
+    consolidated_metadata: defaultdict[str, dict[str, Any]] = defaultdict(dict)
+    for k, v in v2_metadata.items():
+        path, kind = k.rsplit("/.", 1)
+        if kind == "zarray":
+            consolidated_metadata[path].update(v)
+        elif kind == "zattrs":
+            consolidated_metadata[path]["attributes"] = v
+        elif kind == "zgroup":
+            consolidated_metadata[path].update(v)
         else:
-            zmeta = buffer_to_json_object(zgroup_bytes)
+            raise ValueError(f"Invalid file type '{kind}' at path '{path}")
+    return ConsolidatedMetadata.from_dict(
+        {"metadata": dict(consolidated_metadata), "kind": "inline", "must_understand": False}
+    )
+
+
+async def _read_metadata_v3(store: Store, path: str) -> ArrayV3Metadata | GroupMetadata:
+    """`read_v3_metadata`, raising FileNotFoundError instead of returning None."""
+    metadata = await read_v3_metadata(store, path)
+    if metadata is None:
+        raise FileNotFoundError(path)
+    return metadata
+
 
-    return _build_metadata_v2(zmeta, zattrs)
+async def _read_metadata_v2(store: Store, path: str) -> ArrayV2Metadata | GroupMetadata:
+    """`read_v2_metadata`, raising FileNotFoundError instead of returning None."""
+    metadata = await read_v2_metadata(store, path)
+    if metadata is None:
+        raise FileNotFoundError(path)
+    return metadata
 
 
 async def _read_group_metadata_v2(store: Store, path: str) -> GroupMetadata:
@@ -3576,16 +3586,17 @@ def _build_metadata_v3(zarr_json: dict[str, JSON]) -> ArrayV3Metadata | GroupMet
     """
     if "node_type" not in zarr_json:
         msg = "Required key 'node_type' is missing from the provided metadata document."
-        raise MetadataValidationError(msg)
+        raise NodeTypeValidationError(msg)
     match zarr_json:
         case {"node_type": "array"}:
             return ArrayV3Metadata.from_dict(zarr_json)
         case {"node_type": "group"}:
             return GroupMetadata.from_dict(zarr_json)
+        case {"node_type": node_type}:
+            msg = f"Invalid value for 'node_type'. Expected 'array' or 'group'. Got {node_type!r}."
+            raise NodeTypeValidationError(msg)
         case _:  # pragma: no cover
-            raise ValueError(
-                "invalid value for `node_type` key in metadata document"
-            )  # pragma: no cover
+            raise AssertionError("unreachable")  # pragma: no cover
 
 
 def _build_metadata_v2(
@@ -3602,72 +3613,47 @@ def _build_metadata_v2(
 
 
 @overload
-def _build_node(*, store: Store, path: str, metadata: ArrayV2Metadata) -> AsyncArrayV2: ...
+def _build_node(
+    *, store: Store, path: str, metadata: ArrayV2Metadata, config: ArrayConfigLike | None = None
+) -> AsyncArrayV2: ...
 
 
 @overload
-def _build_node(*, store: Store, path: str, metadata: ArrayV3Metadata) -> AsyncArrayV3: ...
+def _build_node(
+    *, store: Store, path: str, metadata: ArrayV3Metadata, config: ArrayConfigLike | None = None
+) -> AsyncArrayV3: ...
 
 
 @overload
-def _build_node(*, store: Store, path: str, metadata: GroupMetadata) -> AsyncGroup: ...
+def _build_node(
+    *, store: Store, path: str, metadata: GroupMetadata, config: ArrayConfigLike | None = None
+) -> AsyncGroup: ...
 
 
 def _build_node(
-    *, store: Store, path: str, metadata: ArrayV3Metadata | ArrayV2Metadata | GroupMetadata
+    *,
+    store: Store,
+    path: str,
+    metadata: ArrayV3Metadata | ArrayV2Metadata | GroupMetadata,
+    config: ArrayConfigLike | None = None,
 ) -> AnyAsyncArray | AsyncGroup:
     """
-    Take a metadata object and return a node (AsyncArray or AsyncGroup).
+    Take a metadata object and return a node (AsyncArray or AsyncGroup). `config`
+    applies to an array and is ignored for a group.
     """
     store_path = StorePath(store=store, path=path)
     match metadata:
         case ArrayV2Metadata() | ArrayV3Metadata():
-            return AsyncArray(metadata, store_path=store_path)
+            return AsyncArray(metadata, store_path=store_path, config=config)
         case GroupMetadata():
             return AsyncGroup(metadata, store_path=store_path)
         case _:  # pragma: no cover
             raise ValueError(f"Unexpected metadata type: {type(metadata)}")  # pragma: no cover
 
 
-async def _get_node_v2(store: Store, path: str) -> AsyncArrayV2 | AsyncGroup:
-    """
-    Read a Zarr v2 AsyncArray or AsyncGroup from a path in a Store.
-
-    Parameters
-    ----------
-    store : Store
-        The store-like object to read from.
-    path : str
-        The path to the node to read.
-
-    Returns
-    -------
-    AsyncArray | AsyncGroup
-    """
-    metadata = await _read_metadata_v2(store=store, path=path)
-    return _build_node(store=store, path=path, metadata=metadata)
-
-
-async def _get_node_v3(store: Store, path: str) -> AsyncArrayV3 | AsyncGroup:
-    """
-    Read a Zarr v3 AsyncArray or AsyncGroup from a path in a Store.
-
-    Parameters
-    ----------
-    store : Store
-        The store-like object to read from.
-    path : str
-        The path to the node to read.
-
-    Returns
-    -------
-    AsyncArray | AsyncGroup
-    """
-    metadata = await _read_metadata_v3(store=store, path=path)
-    return _build_node(store=store, path=path, metadata=metadata)
-
-
-async def get_node(store: Store, path: str, zarr_format: ZarrFormat) -> AnyAsyncArray | AsyncGroup:
+async def get_node(
+    store: Store, path: str, zarr_format: ZarrFormat | None
+) -> AnyAsyncArray | AsyncGroup:
     """
     Get an AsyncArray or AsyncGroup from a path in a Store.
 
@@ -3677,21 +3663,17 @@ async def get_node(store: Store, path: str, zarr_format: ZarrFormat) -> AnyAsync
         The store-like object to read from.
     path : str
         The path to the node to read.
-    zarr_format : {2, 3}
-        The zarr format of the node to read.
+    zarr_format : {2, 3, None}
+        The zarr format of the node to read, or None to detect it.
 
     Returns
     -------
     AsyncArray | AsyncGroup
     """
-
-    match zarr_format:
-        case 2:
-            return await _get_node_v2(store=store, path=path)
-        case 3:
-            return await _get_node_v3(store=store, path=path)
-        case _:  # pragma: no cover
-            raise ValueError(f"Unexpected zarr format: {zarr_format}")  # pragma: no cover
+    metadata = await read_node_metadata(store, path, zarr_format)
+    if metadata is None:
+        raise FileNotFoundError(path)
+    return _build_node(store=store, path=path, metadata=metadata)
 
 
 async def _set_return_key(
diff --git a/src/zarr/core/sync_group.py b/src/zarr/core/sync_group.py
index 8af514e938..c33f65c246 100644
--- a/src/zarr/core/sync_group.py
+++ b/src/zarr/core/sync_group.py
@@ -142,7 +142,7 @@ def create_rooted_hierarchy(
     return _parse_async_node(async_node)
 
 
-def get_node(store: Store, path: str, zarr_format: ZarrFormat) -> AnyArray | Group:
+def get_node(store: Store, path: str, zarr_format: ZarrFormat | None) -> AnyArray | Group:
     """
     Get an Array or Group from a path in a Store.
 
@@ -152,8 +152,8 @@ def get_node(store: Store, path: str, zarr_format: ZarrFormat) -> AnyArray | Gro
         The store-like object to read from.
     path : str
         The path to the node to read.
-    zarr_format : {2, 3}
-        The zarr format of the node to read.
+    zarr_format : {2, 3, None}
+        The zarr format of the node to read, or None to detect it.
 
     Returns
     -------
diff --git a/tests/test_api.py b/tests/test_api.py
index 45d0c0dee4..51aac652a8 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
+import collections
 import inspect
-import re
 from typing import TYPE_CHECKING, Any
 
 import zarr.codecs
@@ -13,9 +13,11 @@
 if TYPE_CHECKING:
     from collections.abc import Callable
     from pathlib import Path
+    from typing import Self
 
-    from zarr.abc.store import Store
-    from zarr.core.common import JSON, MemoryOrder, ZarrFormat
+    from zarr.abc.store import ByteRequest
+    from zarr.core.buffer import Buffer, BufferPrototype
+    from zarr.core.common import JSON, AccessModeLiteral, MemoryOrder, ZarrFormat
     from zarr.types import AnyArray
 
 import contextlib
@@ -30,6 +32,7 @@
 import zarr.api.synchronous
 import zarr.core.group
 from zarr import Array, Group
+from zarr.abc.store import Store
 from zarr.api.synchronous import (
     create,
     create_array,
@@ -42,15 +45,20 @@
     save_array,
     save_group,
 )
-from zarr.core.buffer import NDArrayLike
+from zarr.core.buffer import NDArrayLike, cpu, default_buffer_prototype
 from zarr.errors import (
     ArrayNotFoundError,
+    ContainsArrayError,
+    ContainsGroupError,
     MetadataValidationError,
+    NodeNotFoundError,
+    NodeTypeValidationError,
     ZarrDeprecationWarning,
     ZarrUserWarning,
 )
 from zarr.storage import MemoryStore
 from zarr.storage._utils import normalize_path
+from zarr.storage._wrapper import WrapperStore
 from zarr.testing.utils import gpu_test
 
 
@@ -360,9 +368,9 @@ def test_array_open_array_not_found_sync() -> None:
 def test_v2_and_v3_exist_at_same_path(store: Store) -> None:
     zarr.create_array(store, shape=(10,), dtype="uint8", zarr_format=3)
     zarr.create_array(store, shape=(10,), dtype="uint8", zarr_format=2)
-    msg = f"Both zarr.json (Zarr format 3) and .zarray (Zarr format 2) metadata objects exist at {store}. Zarr v3 will be used."
-    with pytest.warns(ZarrUserWarning, match=re.escape(msg)):
-        zarr.open(store=store)
+    # only zarr.json is read, and the format 3 node is taken without a second look
+    assert zarr.open(store=store).metadata.zarr_format == 3
+    assert zarr.open_array(store=store).metadata.zarr_format == 3
 
 
 @pytest.mark.parametrize("store", ["memory"], indirect=True)
@@ -1377,6 +1385,224 @@ async def test_open_falls_back_to_open_group_async(zarr_format: ZarrFormat) -> N
     assert group.attrs == {"key": "value"}
 
 
+class _CountingStore(WrapperStore[Store]):
+    """A store that records the key of every `get` it forwards."""
+
+    get_counts: collections.Counter[str]
+
+    def __init__(self, store: Store) -> None:
+        super().__init__(store)
+        self.get_counts = collections.Counter()
+
+    def _with_store(self, store: Store) -> Self:
+        # `_with_store` is how a store is re-made read-only, so the copy has to
+        # keep counting into the same tally.
+        new = type(self)(store)
+        new.get_counts = self.get_counts
+        return new
+
+    async def get(
+        self,
+        key: str,
+        prototype: BufferPrototype,
+        byte_range: ByteRequest | None = None,
+    ) -> Buffer | None:
+        self.get_counts[key] += 1
+        return await self._store.get(key, prototype, byte_range)
+
+
+@pytest.mark.filterwarnings("ignore:Consolidated metadata")
+@pytest.mark.parametrize("node", ["array", "group"])
+@pytest.mark.parametrize(
+    ("zarr_format", "use_consolidated"),
+    [(2, None), (2, False), (2, True), (2, "custom"), (3, None), (3, False), (3, True)],
+)
+@pytest.mark.parametrize("mode", ["r", "r+", "a"])
+@pytest.mark.parametrize("path", ["", "parent/child"])
+async def test_open_reads_only_what_the_node_needs(
+    node: Literal["array", "group"],
+    zarr_format: ZarrFormat,
+    use_consolidated: bool | str | None,
+    mode: AccessModeLiteral,
+    path: str,
+) -> None:
+    """`open` reads a format 3 node with one request and a format 2 node with one round of them.
+
+    Whatever it finds has to be what `open_array` or `open_group` would have
+    returned: the same format, path, attributes and read-only-ness, and for a
+    group the same consolidated metadata.
+    """
+    store = _CountingStore(MemoryStore())
+    prefix = f"{path}/" if path else ""
+    if node == "array":
+        await zarr.api.asynchronous.create_array(
+            store,
+            name=path or None,
+            shape=(3,),
+            dtype="uint8",
+            attributes={"key": "value"},
+            zarr_format=zarr_format,
+        )
+    else:
+        await zarr.api.asynchronous.open_group(
+            store, path=path, attributes={"key": "value"}, zarr_format=zarr_format
+        )
+        if use_consolidated:
+            await zarr.api.asynchronous.consolidate_metadata(store, path=path)
+            if isinstance(use_consolidated, str):
+                # move the consolidated document to the non-default key
+                metadata = await store.get(prefix + ".zmetadata", default_buffer_prototype())
+                assert metadata is not None
+                await store.set(prefix + use_consolidated, metadata)
+                await store.delete(prefix + ".zmetadata")
+
+    store.get_counts.clear()
+    result = await zarr.api.asynchronous.open(
+        store=store, path=path, mode=mode, use_consolidated=use_consolidated
+    )
+    assert isinstance(result, AsyncArray if node == "array" else zarr.core.group.AsyncGroup)
+    assert result.metadata.zarr_format == zarr_format
+    assert result.path == path
+    assert result.attrs == {"key": "value"}
+    assert result.store.read_only == (mode == "r")
+    if isinstance(result, zarr.core.group.AsyncGroup):
+        assert (result.metadata.consolidated_metadata is not None) == bool(use_consolidated)
+
+    expected_keys = {"zarr.json"}
+    if zarr_format == 2:
+        expected_keys |= {".zarray", ".zgroup", ".zattrs"}
+        if use_consolidated is not False:
+            expected_keys.add(
+                ".zmetadata" if isinstance(use_consolidated, bool | None) else use_consolidated
+            )
+    assert store.get_counts == {prefix + key: 1 for key in expected_keys}
+
+
+@pytest.mark.parametrize("zarr_format", [2, 3])
+async def test_open_array_does_not_fall_back(zarr_format: ZarrFormat) -> None:
+    """`open` on an array still returns the array rather than falling back to a group."""
+    store = MemoryStore()
+    await zarr.api.asynchronous.create_array(
+        store, shape=(10,), dtype="uint8", zarr_format=zarr_format, attributes={"k": "v"}
+    )
+    arr = await zarr.api.asynchronous.open(store=store)
+    assert isinstance(arr, AsyncArray)
+    assert arr.metadata.zarr_format == zarr_format
+    assert arr.attrs == {"k": "v"}
+
+
+async def test_open_invalid_zarr_format_raises() -> None:
+    """An invalid `zarr_format` is a bad request, not a missing array, so it still raises."""
+    store = MemoryStore()
+    with pytest.raises(
+        MetadataValidationError,
+        match="Invalid value for 'zarr_format'. Expected 2, 3, or None. Got '3.0'.",
+    ):
+        await zarr.api.asynchronous.open(store=store, zarr_format="3.0")  # type: ignore[arg-type]
+
+
+async def test_open_zarr_json_without_node_type_raises() -> None:
+    """A format 3 document must say what it is: no `node_type` is a validation error."""
+    store = MemoryStore()
+    await store.set(
+        "zarr.json", cpu.Buffer.from_bytes(b'{"zarr_format": 3, "attributes": {"k": "v"}}')
+    )
+    with pytest.raises(NodeTypeValidationError, match="Required key 'node_type' is missing"):
+        await zarr.api.asynchronous.open(store=store, mode="r")
+
+
+async def test_open_zarr_json_with_invalid_node_type_raises() -> None:
+    """A `node_type` that is neither array nor group is a validation error, not a missing node."""
+    store = MemoryStore()
+    await store.set("zarr.json", cpu.Buffer.from_bytes(b'{"zarr_format": 3, "node_type": "foo"}'))
+    with pytest.raises(NodeTypeValidationError, match="Expected 'array' or 'group'. Got 'foo'"):
+        await zarr.api.asynchronous.open(store=store, mode="r")
+
+
+@pytest.mark.parametrize("zarr_format", [2, 3])
+async def test_async_array_open_on_group_raises_contains_group(zarr_format: ZarrFormat) -> None:
+    """Opening a group as an array says a group is there, rather than that nothing is."""
+    store = MemoryStore()
+    await zarr.api.asynchronous.open_group(store, zarr_format=zarr_format)
+    with pytest.raises(ContainsGroupError, match="A group exists in store"):
+        await AsyncArray.open(store, zarr_format=zarr_format)
+
+
+@pytest.mark.parametrize("zarr_format", [2, 3])
+async def test_async_group_open_on_array_raises_contains_array(zarr_format: ZarrFormat) -> None:
+    """Opening an array as a group says an array is there, rather than that nothing is."""
+    store = MemoryStore()
+    await zarr.api.asynchronous.create_array(
+        store, shape=(3,), dtype="uint8", zarr_format=zarr_format
+    )
+    with pytest.raises(ContainsArrayError, match="An array exists in store"):
+        await zarr.core.group.AsyncGroup.open(store, zarr_format=zarr_format)
+
+
+@pytest.mark.parametrize("mode", ["r", "a"])
+def test_open_group_on_array_raises_contains_array(mode: AccessModeLiteral) -> None:
+    store = MemoryStore()
+    zarr.create_array(store, shape=(3,), dtype="uint8")
+    with pytest.raises(ContainsArrayError, match="An array exists in store"):
+        zarr.open_group(store, mode=mode)
+
+
+@pytest.mark.parametrize("mode", ["r", "a"])
+def test_open_array_on_group_raises_contains_group(mode: AccessModeLiteral) -> None:
+    store = MemoryStore()
+    zarr.create_group(store)
+    with pytest.raises(ContainsGroupError, match="A group exists in store"):
+        zarr.open_array(store, mode=mode)
+
+
+def _expected_open_outcome(
+    existing: str, mode: str, shape: tuple[int, ...] | None
+) -> type[Array[Any] | Group | Exception]:
+    """The contract of `open`, as a table: what comes back for what is there, the mode, and `shape`."""
+    if mode == "w-" and existing != "nothing":
+        return FileExistsError
+    if mode == "w":
+        return Array if shape else Group
+    if existing == "nothing":
+        if mode in ("r", "r+"):
+            return ArrayNotFoundError if shape else NodeNotFoundError
+        return Array if shape else Group
+    if shape:
+        return Array if existing == "array" else ContainsGroupError
+    return Array if existing == "array" else Group
+
+
+@pytest.mark.parametrize("existing", ["nothing", "array", "group"])
+@pytest.mark.parametrize("mode", ["r", "r+", "a", "w", "w-"])
+@pytest.mark.parametrize("shape", [None, (3,)], ids=["no shape", "shape"])
+def test_open_mode_contract(
+    existing: str, mode: AccessModeLiteral, shape: tuple[int, ...] | None
+) -> None:
+    """`open` follows its two rules for every mode, whatever is at the path, with and without `shape`.
+
+    With `shape` the call describes an array. Without it, the reading modes open
+    whatever node is there and 'a' creates a group when there is none; the
+    creating modes make a group, 'w' over whatever is there and 'w-' only over
+    nothing. An opened node keeps its attributes; a created one has none.
+    """
+    store = MemoryStore()
+    if existing == "array":
+        zarr.create_array(store, shape=(3,), dtype="uint8", attributes={"old": True})
+    elif existing == "group":
+        zarr.create_group(store, attributes={"old": True})
+    kwargs: dict[str, Any] = {} if shape is None else {"shape": shape, "dtype": "uint8"}
+    expected = _expected_open_outcome(existing, mode, shape)
+
+    if issubclass(expected, Exception):
+        with pytest.raises(expected):
+            zarr.open(store=store, mode=mode, **kwargs)
+        return
+    node = zarr.open(store=store, mode=mode, **kwargs)
+    assert isinstance(node, expected)
+    opened = existing != "nothing" and mode in ("r", "r+", "a")
+    assert node.attrs.get("old") is (True if opened else None)
+
+
 @pytest.mark.parametrize("mode", ["r", "r+", "w", "a"])
 def test_open_modes_creates_group(tmp_path: Path, mode: str) -> None:
     # https://github.com/zarr-developers/zarr-python/issues/2490
diff --git a/tests/test_api/test_asynchronous.py b/tests/test_api/test_asynchronous.py
index 6ebec36bbd..8ef9ad0588 100644
--- a/tests/test_api/test_asynchronous.py
+++ b/tests/test_api/test_asynchronous.py
@@ -11,6 +11,7 @@
 from zarr.api.asynchronous import _get_shape_chunks, _like_args, group, open
 from zarr.core.buffer.core import default_buffer_prototype
 from zarr.core.group import AsyncGroup
+from zarr.errors import ContainsGroupError
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -96,19 +97,17 @@ def test_like_args(
     assert _like_args(observed) == expected
 
 
-async def test_open_no_array() -> None:
+async def test_open_with_shape_on_group_raises() -> None:
     """
-    Test that zarr.api.asynchronous.open attempts to open a group when no array is found, but shape was specified in kwargs.
-    This behavior makes no sense but we should still test it.
+    With `shape` given, `open` describes an array, so a group at the path is an error rather than
+    something to fall back to.
     """
     store = {
         "zarr.json": default_buffer_prototype().buffer.from_bytes(
             json.dumps({"zarr_format": 3, "node_type": "group"}).encode("utf-8")
         )
     }
-    with pytest.raises(
-        TypeError, match=r"open_group\(\) got an unexpected keyword argument 'shape'"
-    ):
+    with pytest.raises(ContainsGroupError, match="A group exists in store"):
         await open(store=store, shape=(1,))
 
 
diff --git a/tests/test_group.py b/tests/test_group.py
index 31fbd138cd..5297276a7a 100644
--- a/tests/test_group.py
+++ b/tests/test_group.py
@@ -1547,6 +1547,7 @@ def test_open_mutable_mapping_sync():
 
 
 async def test_open_ambiguous_node():
+    """A path holding both formats opens as Zarr format 3, without reading or warning about the other."""
     zarr_json_bytes = default_buffer_prototype().buffer.from_bytes(
         json.dumps({"zarr_format": 3, "node_type": "group"}).encode("utf-8")
     )
@@ -1554,11 +1555,8 @@ async def test_open_ambiguous_node():
         json.dumps({"zarr_format": 2}).encode("utf-8")
     )
     store: dict[str, Buffer] = {"zarr.json": zarr_json_bytes, ".zgroup": zgroup_bytes}
-    with pytest.warns(
-        ZarrUserWarning,
-        match=r"Both zarr\.json \(Zarr format 3\) and \.zgroup \(Zarr format 2\) metadata objects exist at",
-    ):
-        await AsyncGroup.open(store, zarr_format=None)
+    group = await AsyncGroup.open(store, zarr_format=None)
+    assert group.metadata.zarr_format == 3
 
 
 class TestConsolidated:
diff --git a/tests/test_metadata/test_consolidated.py b/tests/test_metadata/test_consolidated.py
index cd0fd92d74..8f884a0383 100644
--- a/tests/test_metadata/test_consolidated.py
+++ b/tests/test_metadata/test_consolidated.py
@@ -597,16 +597,16 @@ async def test_open_consolidated_raises_async(self, zarr_format: ZarrFormat) ->
     async def v2_consolidated_metadata_empty_dataset(
         self, memory_store: zarr.storage.MemoryStore
     ) -> AsyncGroup:
-        zgroup_bytes = cpu.Buffer.from_bytes(json.dumps({"zarr_format": 2}).encode())
-        zmetadata_bytes = cpu.Buffer.from_bytes(
-            b'{"metadata":{".zgroup":{"zarr_format":2}},"zarr_consolidated_format":1}'
+        await memory_store.set(
+            ".zgroup", cpu.Buffer.from_bytes(json.dumps({"zarr_format": 2}).encode())
         )
-        return AsyncGroup._from_bytes_v2(
-            StorePath(memory_store, path=""),
-            zgroup_bytes,
-            zattrs_bytes=None,
-            consolidated_metadata_bytes=zmetadata_bytes,
+        await memory_store.set(
+            ".zmetadata",
+            cpu.Buffer.from_bytes(
+                b'{"metadata":{".zgroup":{"zarr_format":2}},"zarr_consolidated_format":1}'
+            ),
         )
+        return await AsyncGroup.open(memory_store, zarr_format=2, use_consolidated=True)
 
     async def test_consolidated_metadata_backwards_compatibility(
         self, v2_consolidated_metadata_empty_dataset: AsyncGroup