Merge pull request #76 from dClimate/ci/switch-setup-ipfs-gha

ci(gha): switch to custom gha which allows 0.36 of kubo installation

Author: Faolain

.github/workflows/run-checks.yaml

@@ -37,9 +37,9 @@ jobs:
         run: uv sync
 
       - name: Install IPFS
-        uses: oduwsdl/setup-ipfs@e92fedca9f61ab9184cb74940254859f4d7af4d9 # v0.6.3
+        uses: Faolain/setup-ipfs@v0.7.0
         with:
-          ipfs_version: "0.35.0"
+          ipfs_version: "0.36.0"
           run_daemon: true
 
       - name: Run pytest with coverage

py_hamt/encryption_hamt_store.py

@@ -110,6 +110,19 @@ def __init__(
         self.header = header
         self.metadata_read_cache: dict[str, bytes] = {}
 
+    def with_read_only(self, read_only: bool = False) -> "SimpleEncryptedZarrHAMTStore":
+        if read_only == self.read_only:
+            return self
+
+        clone = type(self).__new__(type(self))
+        clone.hamt = self.hamt
+        clone.encryption_key = self.encryption_key
+        clone.header = self.header
+        clone.metadata_read_cache = self.metadata_read_cache
+        clone._forced_read_only = read_only  # safe; attribute is declared
+        zarr.abc.store.Store.__init__(clone, read_only=read_only)
+        return clone
+
     def _encrypt(self, val: bytes) -> bytes:
         """Encrypts data using ChaCha20-Poly1305."""
         nonce = get_random_bytes(24)  # XChaCha20 uses a 24-byte nonce

py_hamt/store_httpx.py

@@ -1,7 +1,7 @@
 import asyncio
 import re
 from abc import ABC, abstractmethod
-from typing import Any, Dict, Literal, Tuple, cast
+from typing import Any, Literal, Tuple, cast
 
 import httpx
 from dag_cbor.ipld import IPLDKind
@@ -210,27 +210,43 @@ def __init__(
         self.gateway_base_url: str = gateway_base_url
         """@private"""
 
-        self._client_per_loop: Dict[asyncio.AbstractEventLoop, httpx.AsyncClient] = {}
-
         if client is not None:
-            # user supplied → bind it to *their* current loop
-            self._client_per_loop[asyncio.get_running_loop()] = client
-            self._owns_client: bool = False
+            # A client was supplied by the user. We don't own it.
+            self._owns_client = False
+            self._client_per_loop = {asyncio.get_running_loop(): client}
         else:
-            self._owns_client = True  # we'll create clients lazily
+            # No client supplied. We will own any clients we create.
+            self._owns_client = True
+            self._client_per_loop = {}
+
+        # The instance is never closed on initialization.
+        self._closed = False
 
         # store for later use by _loop_client()
         self._default_headers = headers
         self._default_auth = auth
 
         self._sem: asyncio.Semaphore = asyncio.Semaphore(concurrency)
-        self._closed: bool = False
 
     # --------------------------------------------------------------------- #
     # helper: get or create the client bound to the current running loop    #
     # --------------------------------------------------------------------- #
     def _loop_client(self) -> httpx.AsyncClient:
-        """Get or create a client for the current event loop."""
+        """Get or create a client for the current event loop.
+
+        If the instance was previously closed but owns its clients, a fresh
+        client mapping is lazily created on demand.  Users that supplied their
+        own ``httpx.AsyncClient`` still receive an error when the instance has
+        been closed, as we cannot safely recreate their client.
+        """
+        if self._closed:
+            if not self._owns_client:
+                raise RuntimeError("KuboCAS is closed; create a new instance")
+            # We previously closed all internally-owned clients. Reset the
+            # state so that new clients can be created lazily.
+            self._closed = False
+            self._client_per_loop = {}
+
         loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
         try:
             return self._client_per_loop[loop]
@@ -241,7 +257,7 @@ def _loop_client(self) -> httpx.AsyncClient:
                 headers=self._default_headers,
                 auth=self._default_auth,
                 limits=httpx.Limits(max_connections=64, max_keepalive_connections=32),
-                # Uncomment when they finally support Robost HTTP/2 GOAWAY responses
+                # Uncomment when they finally support Robust HTTP/2 GOAWAY responses
                 # http2=True,
             )
             self._client_per_loop[loop] = client
@@ -251,18 +267,20 @@ def _loop_client(self) -> httpx.AsyncClient:
     # graceful shutdown: close **all** clients we own                       #
     # --------------------------------------------------------------------- #
     async def aclose(self) -> None:
-        """Close all internally-created clients."""
-        if not self._owns_client:
-            # User supplied the client; they are responsible for closing it.
+        """
+        Closes all internally-created clients. Must be called from an async context.
+        """
+        if self._owns_client is False:  # external client → caller closes
             return
 
+        # This method is async, so we can reliably await the async close method.
+        # The complex sync/async logic is handled by __del__.
         for client in list(self._client_per_loop.values()):
             if not client.is_closed:
                 try:
                     await client.aclose()
                 except Exception:
-                    # Best-effort cleanup; ignore errors during shutdown
-                    pass
+                    pass  # best-effort cleanup
 
         self._client_per_loop.clear()
         self._closed = True
@@ -277,23 +295,44 @@ async def __aexit__(self, *exc: Any) -> None:
 
     def __del__(self) -> None:
         """Best-effort close for internally-created clients."""
+        if not hasattr(self, "_owns_client") or not hasattr(self, "_closed"):
+            return
+
         if not self._owns_client or self._closed:
             return
 
         # Attempt proper cleanup if possible
         try:
             loop = asyncio.get_running_loop()
         except RuntimeError:
-            loop = None
+            # No running loop - can't do async cleanup
+            # Just clear the client references synchronously
+            if hasattr(self, "_client_per_loop"):
+                # We can't await client.aclose() without a loop,
+                # so just clear the references
+                self._client_per_loop.clear()
+                self._closed = True
+            return
 
+        # If we get here, we have a running loop
         try:
-            if loop is None or not loop.is_running():
-                asyncio.run(self.aclose())
-            else:
+            if loop.is_running():
+                # Schedule cleanup in the existing loop
                 loop.create_task(self.aclose())
+            else:
+                # Loop exists but not running - try asyncio.run
+                coro = self.aclose()  # Create the coroutine
+                try:
+                    asyncio.run(coro)
+                except Exception:
+                    # If asyncio.run fails, we need to close the coroutine properly
+                    coro.close()  # This prevents the RuntimeWarning
+                    raise  # Re-raise to hit the outer except block
         except Exception:
-            # Suppress all errors during interpreter shutdown or loop teardown
-            pass
+            # If all else fails, just clear references
+            if hasattr(self, "_client_per_loop"):
+                self._client_per_loop.clear()
+                self._closed = True
 
     # --------------------------------------------------------------------- #
     # save() – now uses the per-loop client                                 #

py_hamt/zarr_hamt_store.py

@@ -51,6 +51,8 @@ class ZarrHAMTStore(zarr.abc.store.Store):
     ```
     """
 
+    _forced_read_only: bool | None = None  # sentinel for wrapper clones
+
     def __init__(self, hamt: HAMT, read_only: bool = False) -> None:
         """
         ### `hamt` and `read_only`
@@ -79,10 +81,36 @@ def __init__(self, hamt: HAMT, read_only: bool = False) -> None:
         """@private"""
 
     @property
-    def read_only(self) -> bool:
-        """@private"""
+    def read_only(self) -> bool:  # type: ignore[override]
+        if self._forced_read_only is not None:  # instance attr overrides
+            return self._forced_read_only
         return self.hamt.read_only
 
+    def with_read_only(self, read_only: bool = False) -> "ZarrHAMTStore":
+        """
+        Return this store (if the flag already matches) or a *shallow*
+        clone that presents the requested read‑only status.
+
+        The clone **shares** the same :class:`~py_hamt.hamt.HAMT`
+        instance; no flushing, network traffic or async work is done.
+        """
+        # Fast path
+        if read_only == self.read_only:
+            return self  # Same mode, return same instance
+
+        # Create new instance with different read_only flag
+        # Creates a *bare* instance without running its __init__
+        clone = type(self).__new__(type(self))
+
+        # Copy attributes that matter
+        clone.hamt = self.hamt  # Share the HAMT
+        clone._forced_read_only = read_only
+        clone.metadata_read_cache = self.metadata_read_cache.copy()
+
+        # Re‑initialise the zarr base class so that Zarr sees the flag
+        zarr.abc.store.Store.__init__(clone, read_only=read_only)
+        return clone
+
     def __eq__(self, other: object) -> bool:
         """@private"""
         if not isinstance(other, ZarrHAMTStore):
@@ -145,6 +173,9 @@ def supports_partial_writes(self) -> bool:
 
     async def set(self, key: str, value: zarr.core.buffer.Buffer) -> None:
         """@private"""
+        if self.read_only:
+            raise Exception("Cannot write to a read-only store.")
+
         if key in self.metadata_read_cache:
             self.metadata_read_cache[key] = value.to_bytes()
         await self.hamt.set(key, value.to_bytes())
@@ -167,6 +198,8 @@ def supports_deletes(self) -> bool:
 
     async def delete(self, key: str) -> None:
         """@private"""
+        if self.read_only:
+            raise Exception("Cannot write to a read-only store.")
         try:
             await self.hamt.delete(key)
             # In practice these lines never seem to be needed, creating and appending data are the only operations most zarrs actually undergo

pyproject.toml

@@ -9,7 +9,7 @@ dependencies = [
     "dag-cbor>=0.3.3",
     "msgspec>=0.18.6",
     "multiformats[full]>=0.3.1.post4",
-    "zarr>=3.0.8",
+    "zarr==3.0.9",
     "pycryptodome>=3.21.0",
 ]
 

tests/test_async.py

@@ -104,26 +104,30 @@ async def test_kubocas_no_running_loop_in_aclose():
     # Create a client in the current loop
     _ = cas._loop_client()
 
-    # Simulate calling aclose when there's no event loop
-    # We'll mock this by calling the method directly
-    import unittest.mock
-
-    # Test the __del__ method with no running loop scenario
+    # Test __del__ behavior when there's no running loop
     with unittest.mock.patch(
         "asyncio.get_running_loop", side_effect=RuntimeError("No running loop")
     ):
-        # This will trigger the exception path in __del__
-        # where it gets a RuntimeError and sets loop = None
+        # This should handle the no-loop case gracefully
         cas.__del__()
 
-    # Now test the normal aclose path with no running loop
+    # Also test aclose directly with no loop
+    # First close it normally
+    await cas.aclose()
+
+    # Create a new instance
+    cas2 = KuboCAS()
+    _ = cas2._loop_client()
+
+    # Now mock no running loop for aclose
     with unittest.mock.patch(
         "asyncio.get_running_loop", side_effect=RuntimeError("No running loop")
     ):
-        await cas.aclose()
+        # The aclose method should handle this gracefully
+        await cas2.aclose()
 
-    # The client references should be cleared
-    assert len(cas._client_per_loop) == 0
+    # Verify cleanup happened
+    assert len(cas2._client_per_loop) == 0
 
 
 @pytest.mark.asyncio