Fix EmbedStore.__getitem__ race condition between index sync and data read

src/blosc2/embed_store.py

@@ -288,17 +288,30 @@ def __setitem__(
 
     def __getitem__(self, key: str) -> blosc2.NDArray | SChunk | blosc2.ObjectArray | blosc2.BatchArray:
         """Retrieve a node from the embed store."""
-        self._sync_metadata()
-        if key not in self._embed_map:
-            raise KeyError(f"Key '{key}' not found in the embed store.")
-        node_info = self._embed_map[key]
-        urlbase = node_info.get("urlbase", None)
+        if self._shared:
+            with self._backing_schunk.holding_lock():
+                self._sync_metadata()
+                if key not in self._embed_map:
+                    raise KeyError(f"Key '{key}' not found in the embed store.")
+                node_info = self._embed_map[key]
+                urlbase = node_info.get("urlbase", None)
+                if not urlbase:
+                    offset = node_info["offset"]
+                    length = node_info["length"]
+                    serialized_data = bytes(self._store[offset : offset + length])
+        else:
+            self._sync_metadata()
+            if key not in self._embed_map:
+                raise KeyError(f"Key '{key}' not found in the embed store.")
+            node_info = self._embed_map[key]
+            urlbase = node_info.get("urlbase", None)
+            if not urlbase:
+                offset = node_info["offset"]
+                length = node_info["length"]
+                serialized_data = bytes(self._store[offset : offset + length])
         if urlbase:
-            urlpath = blosc2.URLPath(node_info["path"], urlbase=urlbase)
-            return blosc2.open(urlpath, mode="r")
-        offset = node_info["offset"]
-        length = node_info["length"]
-        serialized_data = bytes(self._store[offset : offset + length])
+            # Outside the lock: opening a C2Array involves an HTTP round trip
+            return blosc2.open(blosc2.URLPath(node_info["path"], urlbase=urlbase), mode="r")
         # It is safer to copy data here, as the reference to the SChunk may disappear
         # Use from_cframe so we can deserialize either an NDArray or an SChunk
         return blosc2.from_cframe(serialized_data, copy=True)

tests/test_locking.py

@@ -1176,19 +1176,10 @@ def test_embed_store_cross_process_writers(tmp_path):
             keys = list(estore)
             assert "/seed" in keys
             for key in keys[-3:]:
-                try:
-                    node = estore.get(key)  # a concurrent delete cannot happen here
-                    if node is None:
-                        continue
-                    data = node[:]
-                except RuntimeError:
-                    # Same "listed before it is readable" window the None check
-                    # covers, deeper in: the index already carries the key's
-                    # (offset, length) but the backing schunk has not grown to
-                    # cover it, so EmbedStore.__getitem__ raises "Error while
-                    # getting the slice".  Tolerated only while the writers are
-                    # running -- every key is verified strictly once they exit.
+                node = estore.get(key)  # a concurrent delete cannot happen here
+                if node is None:
                     continue
+                data = node[:]
                 assert len(data) == 10
                 nreads += 1
     finally:
@@ -1277,15 +1268,10 @@ def test_dict_store_cross_process_writers(tmp_path):
             for key in keys:
                 if not key.endswith("ext3"):
                     continue
-                try:
-                    node = dstore.get(key)
-                    if node is None:
-                        continue
-                    data = node[:]
-                except RuntimeError:
-                    # See test_embed_store_cross_process_writers: the index
-                    # carries the key before its bytes are readable.
+                node = dstore.get(key)
+                if node is None:
                     continue
+                data = node[:]
                 assert len(data) == 100
                 nreads += 1
     finally: