fix(store): don't close a shared filesystem in FsspecStore.close()

changes/4157.bugfix.md

@@ -0,0 +1,11 @@
+`MemoryStore` now copies buffers as they are written, so it never retains the
+caller's memory. Previously an uncompressed write handed the store a zero-copy
+view of the user's array, and mutating that array afterwards would silently
+rewrite chunks already committed to the store.
+
+Only `MemoryStore` is affected: stores that serialize on write, such as
+`LocalStore` and `ZipStore`, never aliased the caller's memory. Uncompressed
+writes to a `MemoryStore` are correspondingly slower, since the copy that makes
+the stored data independent is now actually performed; compressed writes are
+unchanged. Buffers supplied through the `store_dict` argument remain the
+caller's responsibility and are stored as-is.

lychee.toml

@@ -1,6 +1,9 @@
 # Configuration for the lychee link checker (https://lychee.cli.rs/).
 # Auto-discovered as ./lychee.toml by the lychee GitHub Action.
 
+# Treat redirect status codes as success rather than failures.
+accept = ["200..=299"]
+
 # Files lychee should not scan for links.
 exclude_path = [
   # mkdocs-material theme overrides: hrefs are Jinja expressions like
@@ -17,4 +20,6 @@ exclude = [
   # documentation of a command rather than a reachable link.
   '^https?://0\.0\.0\.0',
   '^https?://(localhost|127\.0\.0\.1)(:\d+)?',
+  # SPEC 0 page times out but is valid.
+  '^https://scientific-python\.org/specs/spec-0000',
 ]

src/zarr/core/buffer/core.py

@@ -45,6 +45,8 @@ def __getitem__(self, key: slice) -> Self: ...
 
     def __setitem__(self, key: slice, value: Any) -> None: ...
 
+    def copy(self) -> Self: ...
+
 
 @runtime_checkable
 class NDArrayLike(Protocol):

src/zarr/storage/_fsspec.py

@@ -3,7 +3,6 @@
 import json
 import warnings
 from contextlib import suppress
-from logging import getLogger
 from typing import TYPE_CHECKING, Any
 
 from packaging.version import parse as parse_version
@@ -19,8 +18,6 @@
 from zarr.errors import ZarrUserWarning
 from zarr.storage._utils import _dereference_path
 
-logger = getLogger(__name__)
-
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator, Iterable
 
@@ -38,26 +35,6 @@
 )
 
 
-async def _close_fs(fs: AsyncFileSystem) -> None:
-    """
-    Best-effort async close of an fsspec async filesystem owned by FsspecStore.
-
-    For filesystems that expose `set_session()` (e.g. s3fs) the underlying
-    aiohttp `ClientSession` is closed explicitly, which prevents
-    "Unclosed client session" `ResourceWarning`s from aiohttp.  For all
-    other filesystem types the call is a no-op (not every implementation
-    manages an HTTP session directly).
-
-    Note that `set_session()` lazily creates a session if none exists yet, so
-    closing a store that never performed any I/O may instantiate a session
-    purely to close it.  This is accepted best-effort behavior; fsspec does not
-    expose a stable, cross-implementation way to test for an existing session.
-    """
-    if hasattr(fs, "set_session"):
-        session = await fs.set_session()
-        await session.close()
-
-
 def _make_async(fs: AbstractFileSystem) -> AsyncFileSystem:
     """Convert a sync FSSpec filesystem to an async FFSpec filesystem
 
@@ -126,6 +103,15 @@ class FsspecStore(Store):
     ZarrUserWarning
         If the file system (fs) was not created with `asynchronous=True`.
 
+    Notes
+    -----
+    Closing the store does not close the underlying filesystem or its network
+    session. fsspec caches and shares filesystem instances across callers, so
+    the store cannot know whether it is the only user, and closing a shared
+    session would break other stores. The filesystem's lifecycle belongs to
+    whoever created it; use fsspec's own tools (e.g. `clear_instance_cache`)
+    to release it.
+
     See Also
     --------
     FsspecStore.from_upath
@@ -152,9 +138,6 @@ def __init__(
         self.fs = fs
         self.path = path
         self.allowed_exceptions = allowed_exceptions
-        # True only when this store created fs itself (from_url / from_mapper with new instance).
-        # Callers who supply their own fs remain responsible for its lifecycle.
-        self._owns_fs: bool = False
 
         if not self.fs.async_impl:
             raise TypeError("Filesystem needs to support async operations.")
@@ -220,17 +203,13 @@ def from_mapper(
         -------
         FsspecStore
         """
-        original_fs = fs_map.fs
-        fs = _make_async(original_fs)
-        store = cls(
+        fs = _make_async(fs_map.fs)
+        return cls(
             fs=fs,
             path=fs_map.root,
             read_only=read_only,
             allowed_exceptions=allowed_exceptions,
         )
-        # _make_async returns a new instance when converting sync→async; own it.
-        store._owns_fs = fs is not original_fs
-        return store
 
     @classmethod
     def from_url(
@@ -272,39 +251,16 @@ def from_url(
         if not fs.async_impl:
             fs = _make_async(fs)
 
-        store = cls(fs=fs, path=path, read_only=read_only, allowed_exceptions=allowed_exceptions)
-        store._owns_fs = True
-        return store
+        return cls(fs=fs, path=path, read_only=read_only, allowed_exceptions=allowed_exceptions)
 
     def with_read_only(self, read_only: bool = False) -> FsspecStore:
         # docstring inherited
-        new_store = type(self)(
+        return type(self)(
             fs=self.fs,
             path=self.path,
             allowed_exceptions=self.allowed_exceptions,
             read_only=read_only,
         )
-        # The derived store shares the same fs. Transfer ownership so the
-        # surviving store closes it, and clear ours to avoid a double-close.
-        # Otherwise the common `from_url(...).with_read_only()` chain would
-        # drop the only owner (the unreferenced source) and leak the session.
-        new_store._owns_fs = self._owns_fs
-        self._owns_fs = False
-        return new_store
-
-    def close(self) -> None:
-        # docstring inherited
-        if self._owns_fs:
-            from zarr.core.sync import sync as zarr_sync
-
-            # Best-effort: a failure to release the session must not block close(),
-            # but log it so a genuine regression in the close path stays observable
-            # rather than silently reverting to the leaking behavior.
-            try:
-                zarr_sync(_close_fs(self.fs))
-            except Exception:
-                logger.debug("Failed to close owned filesystem %r", self.fs, exc_info=True)
-        super().close()
 
     async def clear(self) -> None:
         # docstring inherited

src/zarr/storage/_memory.py

@@ -26,6 +26,18 @@
 logger = getLogger(__name__)
 
 
+def _copy_buffer(value: Buffer) -> Buffer:
+    """Copy `value` so the store does not retain the caller's memory.
+
+    Encoding a chunk can hand the store a zero-copy view of the user's array
+    (an uncompressed write is the common case), and unlike stores that
+    serialize on write, this one keeps whatever it is given alive in a dict.
+    Without this copy a later mutation of the user's array would rewrite
+    chunks already committed to the store.
+    """
+    return type(value).from_array_like(value.as_array_like().copy())
+
+
 class MemoryStore(Store):
     """
     Store for local memory.
@@ -42,6 +54,12 @@ class MemoryStore(Store):
     supports_writes
     supports_deletes
     supports_listing
+
+    Notes
+    -----
+    Writes copy the buffer they are given, so the store never aliases the
+    caller's memory. Buffers passed via `store_dict` are the caller's
+    responsibility and are stored as-is.
     """
 
     supports_writes: bool = True
@@ -117,7 +135,7 @@ def set_sync(self, key: str, value: Buffer) -> None:
             raise TypeError(
                 f"MemoryStore.set(): `value` must be a Buffer instance. Got an instance of {type(value)} instead."
             )
-        self._store_dict[key] = value
+        self._store_dict[key] = _copy_buffer(value)
 
     def delete_sync(self, key: str) -> None:
         self._check_writable()
@@ -178,13 +196,13 @@ async def set(self, key: str, value: Buffer, byte_range: tuple[int, int] | None
             buf[byte_range[0] : byte_range[1]] = value
             self._store_dict[key] = buf
         else:
-            self._store_dict[key] = value
+            self._store_dict[key] = _copy_buffer(value)
 
     async def set_if_not_exists(self, key: str, value: Buffer) -> None:
         # docstring inherited
         self._check_writable()
         await self._ensure_open()
-        self._store_dict.setdefault(key, value)
+        self._store_dict.setdefault(key, _copy_buffer(value))
 
     async def delete(self, key: str) -> None:
         # docstring inherited